Yi|;TUdZddlmZddlZddlZddlZddlmZmZddlm Z m Z m Z m Z ddl mZddlmZmZmZej&d Zd Zd ed <d Zd ed<dZded<dZded<dZd ed<dZded<dddddZded<gdgd gd!d"Zd#ed$<d*d%Zd*d&ZGd'd(Z gd)Z!y)+aRLM Neo-Cortex -- Feedback Collection Pipeline. Collects thumbs-up / thumbs-down / neutral feedback from Telegram callbacks and generic webhooks, converts each interaction into a DPO PreferencePair, and persists pairs to the ``pl_preference_pairs`` PostgreSQL table. Implements Stories 5.01-5.06 of the RLM Neo-Cortex PRD (Module 5). Infrastructure: - PostgreSQL (Elestio) for durable preference-pair storage - Redis (Elestio) for 24-hour interaction-history cache - No SQLite, no local files, no hardcoded secrets VERIFICATION_STAMP Story: 5.01-5.06 Verified By: parallel-builder Verified At: 2026-02-26 Tests: see tests/rlm/test_feedback.py Coverage: 100% ) annotationsN)datetimetimezone)AnyDictListOptional)UUID)FeedbackCollectorProtocolFeedbackSignalPreferencePairzcore.rlm.feedbackpl_preference_pairsstr _TABLE_NAMEzrlm:interaction_INTERACTION_KEY_PREFIXiQint_INTERACTION_TTL_SECONDSd_DPO_MINIMUM_PAIRStelegram_feedback_DEFAULT_ANNOTATORi_FALLBACK_MAX_CHARSzjThank you for reaching out. I'd be happy to help with your booking. Could you please provide more details?zhI appreciate you sharing that with us. Your feedback has been noted and we will look into this promptly.z_That's a great question. Let me get you the right information. Could you clarify what you need?zJThank you for your message. A team member will follow up with you shortly.)booking complaintquestiondefaultzDict[str, str]_FALLBACK_TEMPLATES)bookappointschedulereservavailab)complainunhappy disappointissueproblemwrongbad)whathowwhenwherewhywho?)rrrzDict[str, List[str]]_FALLBACK_KEYWORDSc|jtjD]\}}tfd|Ds|cSy)z8Return the best-matching intent key for the given input.c3&K|]}|v ywN).0kwlowers */mnt/e/genesis-system/core/rlm/feedback.py z!_detect_intent..Ds.rrU{.sr)r9r2itemsany) input_textintentkeywordsr9s @r:_detect_intentrA@sF    E.446 .X. .M c|r|js tdSt|}t|}t|tk\r |dt}|S)zStory 5.06 -- Return a short, contextual fallback response. Never returns an empty string. Always < 500 characters. Varies by detected intent (booking / complaint / question / default). rN)striprrAlenr)r>r?responses r:_generate_fallback_responserGIsS Z--/"9-- J 'F"6*H 8}++001 OrBceZdZUdZeZded< d ddZddZddZ d ddZ dd Z dd Z dd Z dd Zddd ZddZy)FeedbackCollectora^Collects feedback signals and converts them to DPO preference pairs. Story 5.01 -- constructor Story 5.02 -- record_feedback Story 5.03 -- cache_interaction / get_interaction Story 5.04 -- get_pair_count / get_recent_pairs Story 5.05 -- check_dpo_readiness Story 5.06 -- _generate_fallback_response (module-level helper) r TABLE_NAMENc|xstjjd}|s td||_|xstjjd}||_d|_d|_y)aStory 5.01 -- Initialise collector. Parameters ---------- pg_dsn: PostgreSQL DSN string. Falls back to ``DATABASE_URL`` env var. If neither is provided, raises ``ValueError``. redis_url: Optional Redis URL for interaction caching. Falls back to ``REDIS_URL`` env var. If absent, Redis caching is disabled (get_interaction always returns None). DATABASE_URLz>PostgreSQL DSN is required. Pass pg_dsn= or set DATABASE_URL. REDIS_URLN)osenvironget ValueError_pg_dsn _redis_url_pg_pool_redis)selfpg_dsn redis_url resolved_dsnresolved_rediss r:__init__zFeedbackCollector.__init__nsg"?!? P ) #Abjjnn[&A)7"  rBcK|j9 ddl}|j|jd{|_|jS|jS7!#t$r}t d||d}~wwxYww)z:Return the asyncpg connection pool, creating it if needed.Nrz"Failed to create PostgreSQL pool: )rTasyncpg create_poolrR Exception RuntimeError)rVr]excs r: _get_pg_poolzFeedbackCollector._get_pg_poolst ==  X&-&9&9$,,&G G }}t}}!H X"%Gu#MNTWW Xs7 A6"AA AA6A A3A..A33A6cK|jA|jr5 ddlm}|j |jd|_|jS|jS#t $r }t jd|Yd}~yd}~wwxYww)z/Return the Redis client, creating it if needed.NrT)decode_responseszFailed to connect to Redis: %s)rUrS redis.asyncioasynciofrom_urlr_loggerwarning)rVaioredisras r: _get_rediszFeedbackCollector._get_redissr ;; 4?? 0&//RV/W {{t{{ ?E s.B(AB B%B;BBBc K|tjk(ry|j||d{}|r%|jdd}|jdd}n*|xsi}|jdd}|jdd}t |} |tj k(r|} | } n| } |} t || | td|t||jd} |j| d{| S77w)aConvert a user feedback signal into a DPO PreferencePair. Pipeline: 1. Retrieve cached interaction (input + AI output) 2. Generate fallback response for the rejected side 3. Build PreferencePair (POSITIVE: AI chosen; NEGATIVE: AI rejected) 4. Persist pair to ``pl_preference_pairs`` NEUTRAL signals return None without any DB write. Nr> output_textg?)interaction_id tenant_idsignal)r> chosen_outputrejected_output annotator_id confidencemetadata) r NEUTRALget_interactionrPrGPOSITIVErrrname _persist_pair) rVrprorqcontext interactionr> ai_outputctxfallbackchosenrejectedpairs r:record_feedbackz!FeedbackCollector.record_feedbacks" ^++ +!00NKK $r:J# r:I-RCr2J r2I.z: ^,, ,FHF H! $+"0 ^ ++    &&& OLL 's"*C4C0Brrrsrtrujsondumpsrvrnowrutc)rVrpoolsqlconns r:r{zFeedbackCollector._persist_pairs&&(($ &  <<>  T,,""$$!! 4==) X\\*      )      sgC<C"C<C!C<B C' C# C' C<C%C<!C<#C'%C<'C9-C0 .C95C<cK|jd{}|tjdytd|d|}t j ||t ||d}|j||td{y7n7w)zCache an interaction in Redis with a 24-hour TTL. Key: ``rlm:interaction:{tenant_id}:{interaction_id}`` Returns True if cached successfully, False otherwise. Nu,Redis unavailable — interaction not cachedF:)r>rnrpro)exT) rkrhrirrrrsetr)rVrpror>rnrediskeyvalues r:cache_interactionz#FeedbackCollector.cache_interactionsoo'' = NNI J()9+Q~6FG $&Y,   iiU'?i@@@( As"B BA(B ?BB B cK|jd{}|ytd|d|}|j|d{}|y tj|S7G7#tj t f$rYywxYww)zRetrieve a cached interaction. Returns a dict with keys input_text / output_text, or None. Cross-tenant access is prevented by key structure (tenant_id in key). Nr)rkrrPrloadsJSONDecodeError TypeError)rVrprorrraws r:rxz!FeedbackCollector.get_interaction soo'' =()9+Q~6FGIIcN" ; ::c? "( # $$i0  sEBA)BA B A"B B"A>;B=A>>Bc$K|jd{}|j4d{}|jdtd{}t |dcdddd{S7\7E7'7 #1d{7swYyxYww)z5Return exact COUNT(*) of rows in pl_preference_pairs.NSELECT COUNT(*) AS cnt FROM cnt)rbrfetchrowrr)rVrrrows r:get_pair_countz FeedbackCollector.get_pair_count<s&&((<<> # #T (D[M&RSSCs5z? # # #) #S # # # #sfBA3BA5BA;A7A;! B-A9.B5B7A;9B;B B B  BcLK|jd{}dtd}|j4d{}|j||d{}|Dcgc] }t |c}cdddd{S7k7J72cc}w7#1d{7swYyxYww)zReturn the most recently inserted pairs, newest first. Each row dict contains: input_text, chosen_output, rejected_output, annotator_id, confidence, metadata, created_at. Nz SELECT input_text, chosen_output, rejected_output, annotator_id, confidence, metadata, created_at FROM zC ORDER BY created_at DESC LIMIT $1 )rbrrfetchdict)rVlimitrrrrowsrs r:get_recent_pairsz"FeedbackCollector.get_recent_pairsCs &&((  <<> + +TC//D%)*DG* + + +) +/* + + + +sxB$B"B$BB$BBBB.B0 B$<B =B$B$BB B$B!B B!B$c^K|jd{}|j4d{}|jdtd{}t |d}|j dtd{}|j dtdd{}dddd{d}d}D]\} | d} t | tr tj| } n| xsi} | jd d } | d k(r|d z }R| d k(sX|d z }^D cic]} | dt | d} } dkDr||z nd}|dkDr||z nd}|tk\|t||| dS7M777777#1d{7swYxYw#tjtf$ri} YwxYwcc} ww)aCheck whether there are enough pairs for DPO training. Returns ------- dict with keys: ready -- bool: True when pair_count >= 100 pair_count -- int minimum_required -- int (always 100) pct_positive -- float (0.0-1.0) pct_negative -- float (0.0-1.0) annotator_dist -- dict mapping annotator_id -> count NrrzSELECT metadata FROM z*SELECT annotator_id, COUNT(*) AS cnt FROM z GROUP BY annotator_idrrvrqrmryr NEGATIVErtg)ready pair_countminimum_required pct_positive pct_negativeannotator_dist)rbrrrrr isinstancerrrrrrPr)rVrr total_rowtotal signal_rowsannotator_rows pos_count neg_countrmeta_rawmetasigrrrs r:check_dpo_readinessz%FeedbackCollector.check_dpo_readinessYs&&((<<>  T"mm.{m<IYu-.E!% ' }5!K$(::<[MJ()$N     C:H(C(::h/D ~2((8R(Cj Q  "Q   =K* 58C SZ 0* * /4ai E)S .3ai E)S 00 2((,   U)      ,,,i8D* sF-E#F-E&F-E2E)-E2=E,> E2E.E2# F-.E0/"F-F')F- F-F(5/F-&F-)E2,E2.E20F-2F8E; 9FF-F%"F-$F%%F-)NN)rW Optional[str]rXrreturnNone)rr)rz Optional[Any]r5) rpr rorrqr r|Optional[Dict[str, Any]]rzOptional[PreferencePair])rrrr) rpr rorr>rrnrrbool)rpr rorrr)rr)2)rrrzList[Dict[str, Any]])rzDict[str, Any])__name__ __module__ __qualname____doc__rrJ__annotations__r[rbrkrr{rrxrrrr6rBr:rIrI`s"J!!%#'  !    H (-1 <<< < * < " <|4     : " 8#+,> rBrI)rIrGrrrrr)r>rrr)"r __future__rrloggingrNrrtypingrrrr uuidr contractsr r r getLoggerrhrrrrrrrrr2rArGrI__all__r6rBr:rs(#  ',,PP   . / ) S(00 &#&C-C-S|s^ '^FZD,(.w w |  rB