"""Real agentic loop for the Complaint Guidance Agent — no demo script, no skip flags. Design carried over deliberately from the earlier build (see project memory): the harness decides what step/stage a turn is on from facts it tracks itself, and tells the model that — asking the model to self-report its own stage caused it to drift in the original build. What's new here is that every turn is a live gateway call: fact extraction, response phrasing, and three independent verification passes (citation existence, grounding, relevance), all against the real procedures in knowledge_base/ (see that directory's README for what's genuinely esure-sourced vs. deliberately left out). """ import json import re from pydantic import ValidationError from evidence_store import EvidenceStore from model_client import GatewayClient from schemas import TurnResponse _TURN_RESPONSE_TOOL = TurnResponse.tool_schema() _VULN_INDICATORS = re.compile( r"\b(reduced income|short time at work|earning less|job loss|lost (my|his|her) job|" r"bereavement|illness|disabilit|caring responsibilit|domestic abuse|financial abuse|" r"can'?t afford|cannot afford|struggling to pay|low confidence|can'?t understand|" r"difficulty understanding|limited literacy|language need)\b", re.IGNORECASE, ) _AFFIRMATIVE = re.compile( r"\b(yes|yeah|yep|agrees?|agreed|accepts?|accepted|happy for|happy to|ok(ay)? with|" r"consents?|confirms?|confirmed)\b", re.IGNORECASE, ) _NEGATION_WORD = re.compile(r"\b(not|no\b|never|won'?t|don'?t|refus|declin|reject(?!ion))\b", re.IGNORECASE) ESCALATE_TL = "ESCALATE_TL" # Known gap (see knowledge_base/README.md): these targets have no real source content behind # them. Routing here must escalate, never invent a plausible-sounding replacement. _UNRESOLVED_TARGETS = { "PROC-EXPLAIN-CANCELLATION#S1", "PROC-REPRESENT-DD#S1", "PROC-MANUAL-PAYMENT#S1", } _FACTS_TOOL = { "type": "function", "function": { "name": "update_facts", "description": ( "Record which facts the representative has confirmed so far, from their latest " "note. Only set a field if this note actually confirms it; leave others null." ), "parameters": { "type": "object", "properties": { "missed_dd_count": { "type": ["integer", "null"], "description": "How many Direct Debit collections have been rejected this policy year, if stated.", }, "incorrect_info_given": { "type": ["boolean", "null"], "description": "True if account notes confirm a previous advisor incorrectly told the customer the DD would be reapplied despite multiple missed payments.", }, "vulnerability_disclosed": { "type": ["boolean", "null"], "description": "True if the customer has disclosed a vulnerability indicator (reduced income, illness, etc.).", }, "vulnerability_consent": { "type": ["boolean", "null"], "description": "True/false once the customer has answered whether they consent to the vulnerability being recorded.", }, "customer_accepted": { "type": ["boolean", "null"], "description": "True/false once the customer has accepted or rejected the offered resolution.", }, }, }, }, } _GROUNDING_TOOL = { "type": "function", "function": { "name": "grounding_verdict", "description": "Judge whether every factual claim in the response is supported by the provided source material.", "parameters": { "type": "object", "properties": { "supported": {"type": "boolean"}, "reason": {"type": "string"}, }, "required": ["supported", "reason"], }, }, } _RELEVANCE_TOOL = { "type": "function", "function": { "name": "relevance_verdict", "description": "Judge whether the response actually addresses what the representative just said.", "parameters": { "type": "object", "properties": { "responsive": {"type": "boolean"}, "reason": {"type": "string"}, }, "required": ["responsive", "reason"], }, }, } def _step_lookup(procedure, step_id): for step in procedure.get("steps", []): if step["step_id"] == step_id: return step return None def _branch_matches(expr, facts): """Tiny evaluator for the branch grammar actually used in esure-procedures.json: `FIELD OP VALUE [AND FIELD OP VALUE ...]`, OP in (>=, <, ==). """ for clause in expr.split(" AND "): clause = clause.strip() m = re.match(r"(\w+)\s*(>=|<|==)\s*(\w+)", clause) if not m: return False field, op, raw_value = m.groups() actual = facts.get(field) if actual is None: return False value = {"true": True, "false": False}.get(raw_value.lower()) if value is None: value = int(raw_value) if op == ">=" and not (actual >= value): return False if op == "<" and not (actual < value): return False if op == "==" and not (actual == value): return False return True def _resolve_branch(procedure, after_step_id, facts): for branch in procedure.get("branches", []): if branch["after"] == after_step_id and _branch_matches(branch["when"], facts): return branch["goto"] return procedure.get("no_matching_branch", ESCALATE_TL) def _position(store, facts): """Harness-decided position in the one real procedure — never asks the model. Returns one of: ("step", procedure_id, step_id), ("vulnerability", step_id), ("escalate", reason). """ cmp_proc = store.procedure("CMP-DD-REJECT-001") if "vulnerability_disclosed" not in facts: pass # checked below, after S3/S4 — see note on PROC-VULNERABILITY-FLOW below if facts.get("missed_dd_count") is None: if "missed_dd_count" not in facts: return ("step", "CMP-DD-REJECT-001", "S1") return ("step", "CMP-DD-REJECT-001", "S2") if facts.get("incorrect_info_given") is None: return ("step", "CMP-DD-REJECT-001", "S3") goto = _resolve_branch(cmp_proc, "S3", facts) if goto in _UNRESOLVED_TARGETS: return ("escalate", "No real procedure content for this path yet (see " "knowledge_base/README.md's Known gap section) — hold and escalate to a " "team leader rather than improvise.") if goto == ESCALATE_TL: return ("escalate", "Procedure has no matching branch for these facts.") # goto == "CMP-DD-REJECT-001#S4" — the explain step itself says "end by checking for # vulnerability", so a disclosed-but-unconfirmed vulnerability is handled as an interrupt # here before S5, exactly like PROC-VULNERABILITY-FLOW's own precedence_band says it # should take priority. Once V3 has actually run and set vulnerability_confirmed, the # interrupt is resolved and must not re-trigger — without this check turn 6 of the real # end-to-end test looped back into V3 forever instead of ever reaching S5. if facts.get("vulnerability_disclosed") and "vulnerability_confirmed" not in facts: if facts.get("vulnerability_consent") is None: return ("vulnerability", "V2") return ("vulnerability", "V3") if "incorrect_info_given" in facts and facts.get("_explained") is not True: return ("step", "CMP-DD-REJECT-001", "S4") if facts.get("customer_accepted") is None: return ("step", "CMP-DD-REJECT-001", "S5") goto2 = _resolve_branch(cmp_proc, "S5", facts) if goto2 == ESCALATE_TL: return ("escalate", "Customer rejected the offered resolution.") if not facts.get("_logged"): return ("step", "CMP-DD-REJECT-001", "S6") # S7: case-note posting (see knowledge_base/README.md — added for demo parity with the # original script, not itself from an esure source document). return ("step", "CMP-DD-REJECT-001", "S7") class ReasoningAgent: def __init__(self, client=None, store=None): self.client = client or GatewayClient() self.store = store or EvidenceStore() def run(self, question, history=None, carried=None): history = history or [] facts = dict(carried or {}) if _VULN_INDICATORS.search(question or ""): facts.setdefault("vulnerability_disclosed", True) position = _position(self.store, facts) facts = self._extract_facts(question, history, facts) # Deterministic safety net for the two binary yes/no confirmations the procedure # waits on: Groq's inference occasionally (observed directly, not hypothetical, even # at temperature=0 — a known characteristic of its fast-inference hardware) misses an # unambiguous "yes" and leaves the fact unset, stalling the conversation at the same # step forever. Only applied when the position is actually waiting on that exact fact, # and only for genuinely unambiguous short confirmations — never overrides a "no". if (position[0] == "vulnerability" and position[1] == "V2" and facts.get("vulnerability_consent") is None and _AFFIRMATIVE.search(question or "") and not _NEGATION_WORD.search(question or "")): facts["vulnerability_consent"] = True if (position == ("step", "CMP-DD-REJECT-001", "S5") and facts.get("customer_accepted") is None and _AFFIRMATIVE.search(question or "") and not _NEGATION_WORD.search(question or "")): facts["customer_accepted"] = True # Facts can flip the position (e.g. this note just supplied missed_dd_count) — # recompute once, after applying what this turn actually confirmed. position = _position(self.store, facts) if position[0] == "escalate": return self._escalated_result(position[1], facts, history) instruction, evidence_ids = self._instruction_for(position, facts) context_text = self._context_text(evidence_ids) turn_response = self._phrase_response(question, history, instruction, context_text, evidence_ids) # The step's own instruction IS real source content (it's the procedure's verbatim # text from knowledge_base/esure-procedures.json, not something made up for this # turn) — the grounding judge must check against it too, not just cited documents. # Without this, a turn whose guidance comes entirely from the procedure's own step # text (e.g. S5's "reinstate the mandate, take manual payment") was flagged as # ungrounded purely because that text wasn't separately in context_text — a false # positive found by running this exact scenario end to end. grounding_context = f"PROCEDURE STEP INSTRUCTION FOR THIS TURN:\n{instruction}\n\n{context_text}" verification = self._verify(question, turn_response, grounding_context, evidence_ids, history) if position[0] == "step" and position[2] == "S4": facts["_explained"] = True if position[0] == "step" and position[2] == "S6" and verification["ok"]: facts["_logged"] = True if position[0] == "vulnerability" and position[1] == "V3": facts["vulnerability_confirmed"] = bool(facts.get("vulnerability_consent")) ok = verification["ok"] answer = { "headline": turn_response.headline if ok and turn_response else None, "next_moves": turn_response.next_moves if ok and turn_response else [], "suggested_wording": turn_response.suggested_wording if ok and turn_response else None, "instruction": instruction, "stage": position[1] if position[0] == "vulnerability" else position[2], "citations": [{"reference": aid} for aid in evidence_ids], "case_note": self._case_note(facts) if position == ("step", "CMP-DD-REJECT-001", "S7") else None, } response_json = turn_response.model_dump_json() if turn_response else None return { "answer": answer, "verification": verification, "facts": facts, "turn": len(history) + 1, "_response_json": response_json, } def _escalated_result(self, reason, facts, history): return { "answer": { "headline": "Hold the customer — this needs a team leader.", "next_moves": [reason], "suggested_wording": None, "instruction": reason, "stage": "escalate", "citations": [], "case_note": None, }, "verification": {"ok": True, "citations": [], "grounding": None, "relevance": None, "violations": []}, "facts": facts, "turn": len(history) + 1, "_response_json": None, } def _instruction_for(self, position, facts): kind = position[0] if kind == "vulnerability": step = _step_lookup(self.store.procedure("PROC-VULNERABILITY-FLOW"), position[1]) return step["instruction"], ["VULNERABLE-CUSTOMER-PROCESS"] procedure = self.store.procedure(position[1]) step = _step_lookup(procedure, position[2]) evidence_ids = [] if position[2] == "S5" and facts.get("vulnerability_confirmed"): evidence_ids.append("VULNERABLE-CUSTOMER-PROCESS") return step["instruction"], evidence_ids def _context_text(self, evidence_ids): parts = [] for article_id in evidence_ids: text = self.store.document_text(article_id) if text: parts.append(f"### {article_id}\n{text}") return "\n\n".join(parts) def _extract_facts(self, question, history, facts): # Forced tool_choice isn't supported for this model group through the gateway # (BadRequestError: "does not support forced tool use") — auto plus an explicit # instruction to always call the tool is the fallback the gateway's own error # message recommends. messages = [{ "role": "system", "content": "You extract structured facts from a customer-service representative's " "notes during a missed-Direct-Debit billing complaint call. Only record " "what this specific note actually states. You MUST call the " "update_facts tool on every turn, even if nothing new was confirmed.", }] for turn in history: messages.append({"role": "user", "content": turn.get("note", "")}) messages.append({"role": "user", "content": question}) message = self.client.complete(messages, tools=[_FACTS_TOOL], tool_choice="auto") new_facts = dict(facts) for call in (message.tool_calls or []): if call.function.name != "update_facts": continue try: parsed = json.loads(call.function.arguments) except json.JSONDecodeError: parsed = {} for key, value in parsed.items(): if value is not None: new_facts[key] = value return new_facts def _phrase_response(self, question, history, instruction, context_text, evidence_ids): """Structured, not free prose — see schemas.TurnResponse for why. Forced tool_choice isn't supported by this model group (see _extract_facts), so this is tool_choice="auto" plus an explicit instruction, same pattern used everywhere else in this file. """ system = ( "You are a complaint-guidance assistant helping an esure representative handle a " "live call. You MUST call submit_turn_response with your answer — never reply in " "plain text. Keep headline to one sentence, next_moves to short imperative " "bullets, and base everything ONLY on this instruction and source material — " "never invent a policy detail, number, or process step that isn't in them. This " "includes UI/system details (e.g. 'dropdown list', specific screens), escalation " "routes, verification rituals (e.g. 'read the number back'), or any extra " "precaution/check step (e.g. 'confirm the resolution is still valid before " "closing') that isn't explicitly stated — if the instruction is terse, keep your " "answer just as terse rather than filling in plausible-sounding procedure around " "it, even a well-intentioned one. Only " f"cite ids from this list, never another: {evidence_ids or '(none provided)'}.\n\n" f"Instruction for this turn: {instruction}\n\n" + (f"Source material you may cite or draw on:\n{context_text}\n\n" if context_text else "") ) messages = [{"role": "system", "content": system}] for turn in history: messages.append({"role": "user", "content": turn.get("note", "")}) if turn.get("response_json"): messages.append({"role": "assistant", "content": turn["response_json"]}) messages.append({"role": "user", "content": question}) message = self.client.complete( messages, tools=[_TURN_RESPONSE_TOOL], tool_choice="auto") for call in (message.tool_calls or []): if call.function.name != "submit_turn_response": continue try: return TurnResponse.model_validate_json(call.function.arguments) except ValidationError: return None return None def _verify(self, question, turn_response, context_text, evidence_ids, history): violations = [] if turn_response is None: return {"ok": False, "citations": [], "grounding": None, "relevance": None, "violations": ["model did not return a valid structured response"]} response_text = ( turn_response.headline + "\n" + "\n".join(turn_response.next_moves) + ("\n" + turn_response.suggested_wording if turn_response.suggested_wording else "") ) valid_ids = set(self.store.documents().keys()) for article_id in evidence_ids: if article_id not in valid_ids: violations.append(f"citation references unknown document: {article_id}") # The model declares its own citations too (see schemas.TurnResponse) — it must only # ever cite ids it was actually given, never one it wasn't provided as context. for article_id in turn_response.citations: if article_id not in evidence_ids: violations.append(f"model cited {article_id!r}, which wasn't provided this turn") # Fail-closed: a judge call that doesn't come back with a parsed verdict is treated # as a failed check, not a silent pass — a verification gate that fails open on a # flaky call isn't actually a gate. # # The judge sees the conversation too, not just the cited document — otherwise it # can't tell "an invented policy detail" (a real violation) apart from "a customer # fact the representative already stated" (a legitimate restatement). Found by # running this exact scenario end to end: without conversation context, the judge # flagged the response for saying the customer "is earning less" and gave consent, # both of which the representative had already reported — real false positives. # Include the CURRENT turn's note, not just prior history — without it, the judge # can't see a fact confirmed in this very turn (e.g. consent given right in the # message being responded to), and wrongly flags the response for stating it. conversation = "\n".join(f"- {t.get('note', '')}" for t in history + [{"note": question}]) grounding = None if context_text: grounding = self._judge( _GROUNDING_TOOL, "grounding_verdict", "Does every POLICY/PROCESS claim in this response hold up against the source " "material? The response may also restate facts the representative already " "reported in the conversation (the customer's own circumstances, what they " "said, what was checked) — that is NOT a violation, even though those facts " "aren't in the source material. Only flag a claim about policy, process, or " "company procedure that is NOT supported by the source material. A paraphrase " "is fine; an invented policy/process detail is not.", f"CONVERSATION SO FAR (facts already confirmed, not to be checked against the " f"source):\n{conversation}\n\nSOURCE MATERIAL (check policy/process claims " f"against this):\n{context_text}\n\nRESPONSE:\n{response_text}", ) if not grounding or not grounding.get("supported"): violations.append("grounding check failed: " + (grounding.get("reason", "") if grounding else "no verdict returned")) # Same fix as grounding above: a bare "ready" or "yes" is only interpretable with the # conversation before it — judged alone, it has no visible connection to a response # that (correctly) proceeds with the next documented step. Found by this exact # scenario's turn 7 ("ready" -> post the case note): judged in isolation, the relevance # check called a correct response "unrelated" because it didn't re-explain the context # the representative already knew. relevance = self._judge( _RELEVANCE_TOOL, "relevance_verdict", "Does this response follow sensibly from what the representative just said, GIVEN " "the conversation so far? A short confirmation like \"ready\" or \"yes\" is " "responsive if the response correctly proceeds with the next step implied by the " "conversation — it does not need to restate or re-explain context the " "representative already has.", f"CONVERSATION SO FAR:\n{conversation}\n\nREPRESENTATIVE'S LATEST MESSAGE:\n" f"{question}\n\nRESPONSE:\n{response_text}", ) if not relevance or not relevance.get("responsive"): violations.append("relevance check failed: " + (relevance.get("reason", "") if relevance else "no verdict returned")) return { "ok": not violations, "citations": [{"reference": aid} for aid in evidence_ids], "grounding": grounding, "relevance": relevance, "violations": violations, } def _judge(self, tool, tool_name, system_prompt, user_content): system_prompt = system_prompt + f" You MUST call the {tool_name} tool with your verdict." message = self.client.complete( [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_content}], tools=[tool], tool_choice="auto", ) for call in (message.tool_calls or []): if call.function.name == tool_name: try: return json.loads(call.function.arguments) except json.JSONDecodeError: return None return None def _case_note(self, facts): return ( "Customer complaint:\nCustomer complained after receiving a cancellation notice " f"following {facts.get('missed_dd_count', 'multiple')} missed Direct Debits.\n\n" "Account review:\n" + ("Previous advisor notes confirmed incorrect DD-reapplication advice was given.\n" if facts.get("incorrect_info_given") else "No incorrect prior advice was confirmed.\n") + "\nVulnerability:\n" + (f"Vulnerability disclosed; consent to record: {facts.get('vulnerability_consent')}.\n" if facts.get("vulnerability_disclosed") else "No vulnerability disclosed.\n") + "\nCustomer response:\n" + ("Customer accepted the resolution.\n" if facts.get("customer_accepted") else "Customer response not yet confirmed.\n") )