"""Component 5 — Environment Core (OpenEnv Compatible). Implements the CrimeInvestigationEnv with step/reset/render interface. """ from __future__ import annotations import copy import re from typing import Any, Callable, Optional from crime_env.case_generator import generate_case from crime_env.consistency_tracker import ConsistencyTracker from crime_env.reward_calculator import RewardCalculator from crime_env.agent_prompts import build_system_prompt from crime_env.constants import ( AGENT_NAME_TO_KEY, DETECTIVE_KEY, SUSPECT_A, SUSPECT_A_KEY, SUSPECT_B, SUSPECT_B_KEY, VALID_ACCUSE_TARGETS, VALID_TARGETS, WITNESS_1, WITNESS_1_KEY, ) # ── Action parsing ────────────────────────────────────────────────────────── _ACTION_PATTERNS = { "ask_question": re.compile( r"ACTION:\s*ask_question\s*\|\s*TARGET:\s*(\S+)\s*\|\s*CONTENT:\s*(.+)", re.IGNORECASE, ), "request_evidence": re.compile( r"ACTION:\s*request_evidence\s*\|\s*ITEM:\s*(\S+)", re.IGNORECASE, ), "accuse": re.compile( r"ACTION:\s*accuse\s*\|\s*TARGET:\s*(\S+)", re.IGNORECASE, ), } EVIDENCE_ITEMS = {"keycard_log", "cctv_footage", "forensic_report"} # Evidence confirmation should depend on suspect-specific questioning # that is relevant to the requested evidence type. EVIDENCE_LEAD_TOPICS = { "keycard_log": {"time", "location", "was_at_scene", "alibi"}, "cctv_footage": {"location", "clothing", "time", "was_at_scene"}, "forensic_report": {"clothing", "owns_item", "location", "was_at_scene"}, } # Map evidence item names to indices in physical_evidence list EVIDENCE_INDEX_MAP = { "keycard_log": 0, "cctv_footage": 1, "forensic_report": 2, } TOPIC_KEYWORDS = { "location": [ "east wing", "server room", "vault", "parking lot", "basement archive", "rooftop terrace", "loading dock", "executive suite", "evidence locker", "maintenance corridor", "movie theater", "restaurant", ], "clothing": [ "hoodie", "trench coat", "leather jacket", "overalls", "uniform", "jeans", "windbreaker", "cargo pants", "tracksuit", "bomber jacket", "corduroy", "coat", "jacket", ], "time": [ "7:30", "8:15", "8:45", "9:00", "9:15", "9:45", "10:00", "10:30", "11:15", "11:45", "o'clock", ], "alibi": [ "playing poker", "working late", "at home", "was at", "dinner", "movie", "gym", "volunteering", "party", "sick", "attending", "helping", "hospital", "library", "studying", "babysitting", "coaching", "overtime", "laundromat", "walking", "video call", "prayer", ], "knows_associate": [ "know them", "know him", "know her", "never met", "don't know", "do not know", "know suspect", "knew", ], "was_at_scene": [ "wasn't there", "was not there", "not there", "never been", "was there", "at the scene", "nowhere near", ], "owns_item": [ "not mine", "my bag", "my phone", "my tools", "belongs to me", ], "relationship_to_victim": [ "knew the victim", "never met the victim", "victim", ], } def parse_action(action_string: str) -> dict: """Parse a detective action string into a structured dict. Returns: dict with keys: action_type, target/item, content (if applicable). Returns {action_type: "invalid"} on parse failure. """ action_string = action_string.strip() # Try ask_question m = _ACTION_PATTERNS["ask_question"].search(action_string) if m: return { "action_type": "ask_question", "target": m.group(1).strip(), "content": m.group(2).strip(), } # Try request_evidence m = _ACTION_PATTERNS["request_evidence"].search(action_string) if m: return { "action_type": "request_evidence", "item": m.group(1).strip(), } # Try accuse m = _ACTION_PATTERNS["accuse"].search(action_string) if m: return { "action_type": "accuse", "target": m.group(1).strip(), } return {"action_type": "invalid", "raw": action_string} def extract_topics_from_response(response: str) -> list[tuple[str, str]]: """Extract (topic, value) pairs from a natural language response. Uses word-boundary matching to avoid substring false positives (e.g. 'time' in 'Sometimes', 'own' in 'town'). Returns at most ONE topic per response to prevent a single statement from registering as 3 contradictions simultaneously. Returns: List of (topic, value_snippet) tuples. Max length = 1. """ response_lower = response.lower() # Find the BEST (most specific) single topic match best_match = None best_kw_len = 0 for topic, keywords in TOPIC_KEYWORDS.items(): for kw in keywords: # Use word-boundary regex to avoid substring matches pattern = r'\b' + re.escape(kw) + r'\b' m = re.search(pattern, response_lower) if m: # Prefer longer (more specific) keyword matches if len(kw) > best_kw_len: best_kw_len = len(kw) idx = m.start() start = idx end = min(len(response), idx + len(kw) + 30) value = response[start:end].strip() best_match = (topic, value) break # One match per topic, but continue checking other topics return [best_match] if best_match else [] # ── Environment ───────────────────────────────────────────────────────────── class CrimeInvestigationEnv: """Multi-agent crime investigation RL environment. Compatible with OpenEnv's step/reset/render interface and designed for training with HuggingFace TRL PPO. """ MAX_TURNS = 15 def __init__( self, llm_call: Optional[Callable[[str, list[dict]], str]] = None, ) -> None: """Initialize the environment. Args: llm_call: Callable(system_prompt, conversation_history) -> response. Used for NPC agents (suspects, witness). If None, a simple rule-based fallback is used. """ self.llm_call = llm_call or self._default_llm_call # Will be set on reset() self.case: Optional[dict] = None self.tracker: Optional[ConsistencyTracker] = None self.reward_calc: Optional[RewardCalculator] = None self.turn: int = 0 self.done: bool = False self.conversation_history: list[dict] = [] self.evidence_log: list[dict] = [] self.system_prompts: dict[str, str] = {} self.asked_topics: dict[str, set] = {} # {target: {topics asked}} self._pattern_penalized: set[str] = set() self._pattern_exploited_targets: set[str] = set() # Deflection tracking: stores (suspect, deflection_target) when a # suspect mentions the other suspect in their response. # Only set when the *previous action* was ask_question to a suspect. self._pending_deflection: Optional[tuple[str, str]] = None # What the last action type was (ask_question / request_evidence / etc.) self._last_action_type: Optional[str] = None # Cap anti-deflection bonus to once per suspect per episode. self._deflection_resistance_rewarded: set[str] = set() # Prevent repeated contradiction farming on the same target/topic pair. self._contradiction_rewarded_topics: set[tuple[str, str]] = set() # ── OpenEnv Interface ─────────────────────────────────────────────── def _reset_episode_tracking(self) -> None: self._pending_deflection = None self._last_action_type = None self._pattern_penalized = set() self._pattern_exploited_targets = set() self._deflection_resistance_rewarded = set() self._contradiction_rewarded_topics = set() def reset(self, case_data: Optional[dict] = None) -> dict: """Start a new episode. Returns: Initial observation for the detective agent. """ self.case = copy.deepcopy(case_data) if case_data is not None else generate_case() self.tracker = ConsistencyTracker() self.reward_calc = RewardCalculator() self.turn = 0 self.done = False self.conversation_history = [] self.evidence_log = [] self.asked_topics = {t: set() for t in VALID_TARGETS} self._reset_episode_tracking() # Build system prompts for all agents self.system_prompts = { DETECTIVE_KEY: build_system_prompt("detective", self.case), SUSPECT_A: build_system_prompt(SUSPECT_A_KEY, self.case), SUSPECT_B: build_system_prompt(SUSPECT_B_KEY, self.case), WITNESS_1: build_system_prompt("witness", self.case), } return { "role": "detective", "briefing": self.case["detective_briefing"], "turn": 0, "conversation_history": [], } def step(self, action_string: str) -> tuple[dict, float, bool, dict]: """Execute one detective action. Args: action_string: Action in the format defined in Component 4. Returns: (observation, reward, done, info) """ if self.done: raise RuntimeError("Episode is done. Call reset().") parsed = parse_action(action_string) action_type = parsed["action_type"] if action_type == "ask_question": return self._handle_ask_question(parsed) elif action_type == "request_evidence": return self._handle_request_evidence(parsed) elif action_type == "accuse": return self._handle_accuse(parsed) else: # Invalid action — treat as wasted turn reward = self._apply_invalid_turn_penalty("invalid") if self.turn >= self.MAX_TURNS: return self._handle_timeout(reward) obs = self._build_observation( message="Invalid action format. Please use the correct format." ) return obs, reward, False, {"action": "invalid"} def render(self) -> None: """Print the conversation history in a readable format.""" print("\n" + "=" * 70) print(" CRIME INVESTIGATION — CONVERSATION LOG") print("=" * 70) if self.case: print(f" Crime: {self.case['crime']}") print(f" Location: {self.case['location']}") print("-" * 70) for entry in self.conversation_history: turn = entry.get("turn", "?") speaker = entry.get("speaker", "???") content = entry.get("content", "") flags = entry.get("flags", []) print(f"\n [Turn {turn}] {speaker}:") print(f" {content}") if flags: for flag in flags: print(f" ⚠ FLAG: {flag}") print("\n" + "=" * 70) if self.done and self.reward_calc: rewards = self.reward_calc.get_rewards() print(" FINAL REWARDS:") for agent, r in rewards.items(): print(f" {agent}: {r:+.2f}") print("=" * 70 + "\n") def state(self) -> dict: """Return episode metadata (OpenEnv compatible).""" return { "turn": self.turn, "done": self.done, "max_turns": self.MAX_TURNS, "evidence_revealed": len(self.evidence_log), "contradictions_found": ( len(self.tracker.get_summary()) if self.tracker else 0 ), } # ── Action handlers ───────────────────────────────────────────────── def _handle_ask_question( self, parsed: dict ) -> tuple[dict, float, bool, dict]: target = parsed["target"] question = parsed["content"] step_reward = 0.0 info: dict[str, Any] = {"action": "ask_question", "target": target} flags: list[str] = [] # Validate target if target not in VALID_TARGETS: reward = self._apply_invalid_turn_penalty("ask_question") if self.turn >= self.MAX_TURNS: return self._handle_timeout(reward) obs = self._build_observation( message=f"Invalid target: {target}. Valid: {list(VALID_TARGETS)}" ) return obs, reward, False, {"action": "invalid_target"} step_reward += self._apply_deflection_logic(target) # Per-turn cost self.reward_calc.apply_event("per_turn_cost") step_reward += -0.3 # Check for redundant question (topic already asked to this target) question_topics = extract_topics_from_response(question) if not question_topics: # Bucket unclassified questions to prevent repetition bypasses. question_topics = [("general", question[:40])] question_topic_names = {topic for topic, _ in question_topics} previous_topics = set(self.asked_topics[target]) is_redundant = any(topic in previous_topics for topic in question_topic_names) for topic in question_topic_names: self.asked_topics[target].add(topic) if is_redundant: self.reward_calc.apply_event("redundant_question") step_reward += -0.5 flags.append("redundant_question") # Add question to conversation history self.conversation_history.append( { "turn": self.turn, "speaker": "Detective", "content": f"[To {target}] {question}", "flags": list(flags), } ) # Get NPC response system_prompt = self.system_prompts.get(target, "") # Use full history to avoid artificial contradictions from short memory cap recent_history = list(self.conversation_history) response = self.llm_call(system_prompt, recent_history) response_lower = response.lower() # Extract topics from response and run consistency checks response_topics = extract_topics_from_response(response) response_flags: list[str] = [] step_reward += self._analyse_response_topics( target=target, response_topics=response_topics, question_topic_names=question_topic_names, previous_topics=previous_topics, response_flags=response_flags, ) # Check for witness bias-driven false implication if target == WITNESS_1: self._check_witness_bias(response_lower, response_flags) # Add response to conversation history self.conversation_history.append( { "turn": self.turn, "speaker": target, "content": response, "flags": response_flags, } ) # ── Track deflection attempt ──────────────────────────────────── # Only track deflection when current action is ask_question # to a suspect, and their response mentions the other suspect. if target in (SUSPECT_A, SUSPECT_B): other = SUSPECT_B if target == SUSPECT_A else SUSPECT_A if ( other.lower().replace("_", " ") in response_lower or other.lower() in response_lower ): self._pending_deflection = (target, other) info["deflection_target"] = other else: self._pending_deflection = None else: self._pending_deflection = None self._last_action_type = "ask_question" self.turn += 1 if self.turn >= self.MAX_TURNS: return self._handle_timeout(step_reward) obs = self._build_observation() info["flags"] = response_flags return obs, step_reward, False, info def _apply_deflection_logic(self, target: str) -> float: """Apply deflection rewards/penalties from the previous turn.""" step_delta = 0.0 if ( self._pending_deflection is not None and self._last_action_type == "ask_question" ): deflector, deflection_target = self._pending_deflection if target == deflection_target: deflector_key = AGENT_NAME_TO_KEY.get(deflector, deflector.lower()) self.reward_calc.apply_event( "successful_deflection", agent=deflector_key, ) elif ( target == deflector and deflector not in self._deflection_resistance_rewarded ): # Reward resisting deflection only when the detective stays # on the same suspect, and only once per deflector. self._deflection_resistance_rewarded.add(deflector) step_delta += self.reward_calc.apply_event("deflection_resistance") self._pending_deflection = None return step_delta def _analyse_response_topics( self, target: str, response_topics: list[tuple[str, str]], question_topic_names: set[str], previous_topics: set[str], response_flags: list[str], ) -> float: """Analyze response claims for contradictions, lies, and prior-pattern hits.""" step_delta = 0.0 agent_key = AGENT_NAME_TO_KEY.get(target, target.lower()) for topic, value in response_topics: claim_result = self.tracker.record_claim(target, topic, value, self.turn) if claim_result["contradicted"]: response_flags.append( f"CONTRADICTION on '{topic}': " f"previously said '{claim_result['old_value'][:50]}', " f"now says '{claim_result['new_value'][:50]}'" ) intentional_probe = ( topic in question_topic_names and topic in previous_topics ) contradiction_key = (target, topic) if ( intentional_probe and contradiction_key not in self._contradiction_rewarded_topics ): self._contradiction_rewarded_topics.add(contradiction_key) self.reward_calc.apply_event("contradiction_exposed") step_delta += 2.0 if target in (SUSPECT_A, SUSPECT_B): self.reward_calc.apply_event( "suspect_self_contradiction", agent=agent_key, ) elif target == WITNESS_1: self.reward_calc.apply_event("witness_self_contradiction") visible_evidence = list(self.evidence_log) if visible_evidence: ev_result = self.tracker.check_against_evidence( target, topic, value, visible_evidence ) if ev_result["confirmed_lie"]: response_flags.append( f"LIE CONFIRMED by evidence on '{topic}'" ) if target in (SUSPECT_A, SUSPECT_B): self.reward_calc.apply_event( "confirmed_lie_vs_evidence", agent=agent_key, ) step_delta += 1.0 prior_result = self.tracker.check_prior_pattern( target, topic, value, self.case["prior_history"] ) if prior_result["pattern_match"]: response_flags.append( f"PRIOR PATTERN MATCH on '{topic}'" ) has_documented_history = bool( self.case["prior_history"] .get(target, {}) .get("past_lies_on_record", []) ) topic_aligned_probe = topic in question_topic_names if has_documented_history and topic_aligned_probe: if target not in self._pattern_exploited_targets: self._pattern_exploited_targets.add(target) self.reward_calc.apply_event("prior_pattern_exploited") step_delta += 1.5 if ( target == self.case["criminal"] and target not in self._pattern_penalized ): self._pattern_penalized.add(target) self.reward_calc.apply_event( "prior_pattern_repeated", agent=agent_key, ) return step_delta def _apply_invalid_turn_penalty(self, action_type: str) -> float: """Apply consistent malformed-action accounting across all handlers.""" self.turn += 1 self.reward_calc.apply_event("per_turn_cost") self._last_action_type = action_type self._pending_deflection = None return -0.3 def _check_witness_bias(self, response_lower: str, response_flags: list[str]) -> None: """Flag likely witness-bias implication of the innocent suspect.""" witness_knowledge = self.case["agent_knowledge"][WITNESS_1_KEY] if not witness_knowledge.get("bias_target"): return bias_target = witness_knowledge["bias_target"] # Only fire bias penalty if the bias target is the INNOCENT suspect. if bias_target.lower() not in response_lower or bias_target == self.case["criminal"]: return bias_strength = witness_knowledge.get("bias_strength", 0) if bias_strength <= 0.2: return scene_location = self.case["location"].lower() bias_name = bias_target.lower() implication_patterns = [ f"saw {bias_name}", f"{bias_name} was", f"{bias_name} at", f"{bias_name} near", f"{bias_name} did it", f"{bias_name} committed", ] certainty_markers = [ "definitely", "certain", "for sure", "i'm sure", "it was", ] uncertainty_markers = [ "not sure", "might", "maybe", "could", "possibly", "can't be certain", "uncertain", ] implies_innocent = any(p in response_lower for p in implication_patterns) mentions_scene = ( scene_location in response_lower or "scene" in response_lower or "there" in response_lower ) is_certain = any(m in response_lower for m in certainty_markers) is_uncertain = any(m in response_lower for m in uncertainty_markers) if implies_innocent and mentions_scene and (is_certain or "saw" in response_lower) and not is_uncertain: self.reward_calc.apply_event("bias_driven_false_implication") response_flags.append("POSSIBLE BIAS-DRIVEN STATEMENT") def _handle_request_evidence( self, parsed: dict ) -> tuple[dict, float, bool, dict]: item = parsed["item"].lower() step_reward = 0.0 info: dict[str, Any] = {"action": "request_evidence", "item": item} # Evidence request clears any pending deflection — it's not a follow-up self._pending_deflection = None self._last_action_type = "request_evidence" self.reward_calc.apply_event("per_turn_cost") step_reward += -0.3 self.turn += 1 if item not in EVIDENCE_ITEMS: if self.turn >= self.MAX_TURNS: return self._handle_timeout(step_reward) obs = self._build_observation( message=f"Unknown evidence item: {item}. " f"Available: {list(EVIDENCE_ITEMS)}" ) return obs, step_reward, False, info # Check if this evidence was already revealed (prevent reward hacking) already_revealed = any( e.get("name") == item for e in self.evidence_log ) if already_revealed: if self.turn >= self.MAX_TURNS: return self._handle_timeout(step_reward) obs = self._build_observation( message=f"Evidence '{item}' has already been revealed." ) return obs, step_reward, False, info # Reveal the evidence idx = EVIDENCE_INDEX_MAP.get(item, 0) evidence = copy.deepcopy(self.case["physical_evidence"][idx]) evidence["visible_to_detective"] = True self.case["physical_evidence"][idx]["visible_to_detective"] = True self.evidence_log.append(evidence) # Check if it confirms an existing lead points_to = evidence.get("points_to", "unknown") if points_to != "unknown": # Only grant reward if the detective asked evidence-relevant, # non-general questions to the implicated suspect. if points_to in VALID_TARGETS: asked = self.asked_topics.get(points_to, set()) lead_topics = EVIDENCE_LEAD_TOPICS.get(item, set()) has_meaningful_lead = any( topic != "general" and topic in lead_topics for topic in asked ) else: has_meaningful_lead = False if has_meaningful_lead: self.reward_calc.apply_event("evidence_request_confirms_lead") step_reward += 1.0 info["confirms_lead"] = points_to # IMPORTANT: Do NOT reveal points_to to the detective — they must reason self.conversation_history.append( { "turn": self.turn - 1, "speaker": "Evidence", "content": f"[{item}] {evidence['description']}", "flags": [], } ) if self.turn >= self.MAX_TURNS: return self._handle_timeout(step_reward) obs = self._build_observation( message=f"Evidence revealed: {evidence['description']}" ) info["evidence"] = evidence return obs, step_reward, False, info def _handle_accuse( self, parsed: dict ) -> tuple[dict, float, bool, dict]: target = parsed["target"] if target not in VALID_ACCUSE_TARGETS: reward = self._apply_invalid_turn_penalty("accuse") if self.turn >= self.MAX_TURNS: return self._handle_timeout(reward) obs = self._build_observation( message=( f"Invalid accusation target: {target}. " f"Valid: {list(VALID_ACCUSE_TARGETS)}" ) ) return obs, reward, False, { "action": "invalid_accuse_target", "target": target, } self.done = True criminal = self.case["criminal"] correct = target == criminal info: dict[str, Any] = { "action": "accuse", "target": target, "correct": correct, "actual_criminal": criminal, } # Check if witness bias was active witness_knowledge = self.case["agent_knowledge"][WITNESS_1_KEY] bias_active = ( witness_knowledge.get("bias_target") is not None and witness_knowledge.get("bias_strength", 0) > 0.2 ) if correct: # Correct accusation self.reward_calc.apply_event("correct_accusation") # Criminal caught criminal_key = AGENT_NAME_TO_KEY.get(criminal, criminal.lower()) self.reward_calc.apply_event( "correctly_caught", agent=criminal_key ) # Witness helped correctly self.reward_calc.apply_event("helped_correct_accusation") else: # Wrong accusation self.reward_calc.apply_event( "wrong_accusation", bias_active=bias_active ) # The real criminal escapes criminal_key = AGENT_NAME_TO_KEY.get(criminal, criminal.lower()) self.reward_calc.apply_event( "escaped_not_accused", agent=criminal_key ) # Innocent suspect was wrongly accused innocent_key = AGENT_NAME_TO_KEY.get(target, target.lower()) self.reward_calc.apply_event( "wrongly_accused_innocent", agent=innocent_key ) # Witness led to wrong accusation self.reward_calc.apply_event("led_to_wrong_accusation") self.conversation_history.append( { "turn": self.turn, "speaker": "Detective", "content": f"ACCUSATION: {target}", "flags": [ "✅ CORRECT" if correct else "❌ WRONG", ], } ) # Bug 1: Return only the terminal delta, not cumulative total if correct: terminal_delta = 10.0 else: terminal_delta = -8.0 if not bias_active else -10.0 obs = self._build_observation( message=( f"Accusation: {target}. " f"{'CORRECT!' if correct else 'WRONG!'} " f"The criminal was {criminal}." ) ) return obs, terminal_delta, True, info def _handle_timeout(self, step_reward: float = 0.0) -> tuple[dict, float, bool, dict]: """Handle the case where 15 turns pass with no accusation.""" self.done = True self.reward_calc.apply_event("timeout_no_accusation") # Criminal escapes criminal = self.case["criminal"] criminal_key = AGENT_NAME_TO_KEY.get(criminal, criminal.lower()) self.reward_calc.apply_event( "escaped_not_accused", agent=criminal_key ) self.conversation_history.append( { "turn": self.turn, "speaker": "System", "content": "TIMEOUT — No accusation made. Case goes cold.", "flags": ["⏰ TIMEOUT"], } ) # Bug 3: Return terminal delta (-3.0) instead of cumulative total timeout_penalty = -3.0 obs = self._build_observation( message="Case timed out. No accusation was made within 15 turns." ) return obs, step_reward + timeout_penalty, True, {"action": "timeout"} # ── Helpers ────────────────────────────────────────────────────────── def _build_observation(self, message: str = "") -> dict: """Build the observation dict for the detective.""" return { "role": "detective", "briefing": self.case["detective_briefing"], "turn": self.turn, "conversation_history": list(self.conversation_history), # Bug 10: Deep copy evidence dicts to prevent mutation of case state "evidence_log": copy.deepcopy(self.evidence_log), "message": message, } @staticmethod def _default_llm_call( system_prompt: str, conversation_history: list[dict] ) -> str: """Rule-based fallback for NPC agents. Uses the system prompt to identify the agent and provide role-appropriate responses that stay consistent. Works with either suspect as the criminal. """ # Determine role from system prompt # Bug 5: Use 'FAKE ALIBI' to identify guilty suspect instead of 'committed' # (detective prompt contains 'committed' too, causing misidentification) is_suspect = "Suspect_A" in system_prompt or "Suspect_B" in system_prompt is_guilty = "FAKE ALIBI:" in system_prompt is_innocent = "REAL ALIBI:" in system_prompt is_witness = ("Witness_1" in system_prompt or "Witness" in system_prompt) and not is_suspect if is_suspect and is_guilty: # Extract alibi from the prompt alibi = "at a family gathering" if "FAKE ALIBI:" in system_prompt: alibi_line = system_prompt.split("FAKE ALIBI:")[1].split("\n")[0].strip() if alibi_line: alibi = alibi_line return ( f"I was nowhere near the scene. {alibi}. " "Check with anyone who was there — they'll confirm. " "I don't see why you're asking me about this." ) elif is_suspect and is_innocent: # Extract alibi from the prompt alibi = "working late at the office" if "REAL ALIBI:" in system_prompt: alibi_line = system_prompt.split("REAL ALIBI:")[1].split("\n")[0].strip() if alibi_line: alibi = alibi_line return ( f"I can tell you exactly where I was — {alibi}. " "I have receipts and witnesses to prove it. " "I had absolutely nothing to do with this." ) elif is_witness: # Extract what the witness saw from the prompt saw = "someone in dark clothing near the area" if "WHAT YOU SAW:" in system_prompt: saw_line = system_prompt.split("WHAT YOU SAW:")[1].split("\n")[0].strip() if saw_line: saw = saw_line return ( f"I saw {saw}. " "They were moving quickly and seemed nervous. " "I didn't get a clear look at their face." ) else: return "I don't have anything else to add."