scripts/conversation_depth_analyzer.py
7.2 KB| 1 | #!/usr/bin/env python3 |
| 2 | """conversation_depth_analyzer.py — Detect implicit reflect-trigger signals. |
| 3 | |
| 4 | Stdlib-only. Analyzes a conversation transcript and reports: |
| 5 | |
| 6 | - turn count (User: + Assistant: pairs) |
| 7 | - detail-mode turns (turns dominated by implementation specifics) |
| 8 | - frustration markers (signs of user stuck-ness) |
| 9 | - dead-end signals (pivots within short span) |
| 10 | - implicit-trigger verdict: whether the conversation matches reflect-skill auto-invocation criteria |
| 11 | |
| 12 | The skill OFFERS reflection when implicit signals fire; it does NOT auto-invoke. |
| 13 | |
| 14 | NO LLM CALLS. Pure regex + counting. |
| 15 | |
| 16 | Usage: |
| 17 | python conversation_depth_analyzer.py --conversation /tmp/transcript.txt |
| 18 | python conversation_depth_analyzer.py --conversation /tmp/transcript.txt --output json |
| 19 | python conversation_depth_analyzer.py --sample |
| 20 | """ |
| 21 | |
| 22 | import argparse |
| 23 | import json |
| 24 | import re |
| 25 | import sys |
| 26 | from pathlib import Path |
| 27 | from typing import Any, Dict, List |
| 28 | |
| 29 | |
| 30 | TURN_RE = re.compile(r"^\s*(User|Assistant):\s*", re.MULTILINE) |
| 31 | |
| 32 | DETAIL_MARKERS = [ |
| 33 | r"\bimplementation\b", |
| 34 | r"\bcode\b", |
| 35 | r"\bfunction\b", |
| 36 | r"\bclass\b", |
| 37 | r"\bvariable\b", |
| 38 | r"\b(syntax|method|parameter|argument)\b", |
| 39 | r"\bdebug\b", |
| 40 | r"\berror\b", |
| 41 | r"`[^`]+`", # backtick-quoted code references |
| 42 | ] |
| 43 | |
| 44 | FRUSTRATION_MARKERS = [ |
| 45 | r"\b(ugh|argh|frustrated|stuck)\b", |
| 46 | r"\bnot working\b", |
| 47 | r"\bdoesn'?t work\b", |
| 48 | r"\bgoing in circles\b", |
| 49 | r"\bstill (broken|failing|wrong)\b", |
| 50 | r"\bwhy isn'?t\b", |
| 51 | r"\bthis is (weird|strange|odd|confusing)\b", |
| 52 | ] |
| 53 | |
| 54 | DEAD_END_MARKERS = [ |
| 55 | r"\bnope\b", |
| 56 | r"\bthat didn'?t work\b", |
| 57 | r"\b(let's|let me) try (something|a) (else|different)\b", |
| 58 | r"\bback to\b", |
| 59 | r"\bnever mind\b", |
| 60 | r"\bscratch that\b", |
| 61 | ] |
| 62 | |
| 63 | |
| 64 | def count_turns(text: str) -> Dict[str, int]: |
| 65 | matches = TURN_RE.findall(text) |
| 66 | user_turns = sum(1 for m in matches if m == "User") |
| 67 | assistant_turns = sum(1 for m in matches if m == "Assistant") |
| 68 | return { |
| 69 | "total_turns": len(matches), |
| 70 | "user_turns": user_turns, |
| 71 | "assistant_turns": assistant_turns, |
| 72 | } |
| 73 | |
| 74 | |
| 75 | def count_pattern_hits(text: str, patterns: List[str]) -> int: |
| 76 | return sum(len(re.findall(p, text, re.IGNORECASE)) for p in patterns) |
| 77 | |
| 78 | |
| 79 | def detect_detail_mode_run(text: str) -> int: |
| 80 | """Count consecutive turns that have detail markers but no strategic check-in.""" |
| 81 | blocks = TURN_RE.split(text) |
| 82 | consecutive_detail = 0 |
| 83 | max_consecutive = 0 |
| 84 | for block in blocks: |
| 85 | if not block.strip(): |
| 86 | continue |
| 87 | if any(re.search(p, block, re.IGNORECASE) for p in DETAIL_MARKERS): |
| 88 | consecutive_detail += 1 |
| 89 | max_consecutive = max(max_consecutive, consecutive_detail) |
| 90 | else: |
| 91 | consecutive_detail = 0 |
| 92 | return max_consecutive |
| 93 | |
| 94 | |
| 95 | def analyze(text: str) -> Dict[str, Any]: |
| 96 | turns = count_turns(text) |
| 97 | detail_mode_max = detect_detail_mode_run(text) |
| 98 | frustration_count = count_pattern_hits(text, FRUSTRATION_MARKERS) |
| 99 | dead_end_count = count_pattern_hits(text, DEAD_END_MARKERS) |
| 100 | |
| 101 | signals: List[str] = [] |
| 102 | if detail_mode_max >= 5: |
| 103 | signals.append(f"Detail-mode run of {detail_mode_max} consecutive turns") |
| 104 | if turns["total_turns"] >= 20: |
| 105 | signals.append(f"Long conversation: {turns['total_turns']} turns total") |
| 106 | if frustration_count >= 2: |
| 107 | signals.append(f"Frustration markers: {frustration_count} occurrences") |
| 108 | if dead_end_count >= 2: |
| 109 | signals.append(f"Dead-end / pivot markers: {dead_end_count} occurrences") |
| 110 | |
| 111 | should_offer_reflect = len(signals) >= 2 or detail_mode_max >= 10 |
| 112 | |
| 113 | return { |
| 114 | "turns": turns, |
| 115 | "max_consecutive_detail_turns": detail_mode_max, |
| 116 | "frustration_hits": frustration_count, |
| 117 | "dead_end_hits": dead_end_count, |
| 118 | "trigger_signals": signals, |
| 119 | "implicit_trigger_verdict": ( |
| 120 | "OFFER_REFLECT" if should_offer_reflect else "NO_TRIGGER" |
| 121 | ), |
| 122 | "note": ( |
| 123 | "Skill should OFFER reflection (ask user if they want to step back), not auto-invoke" |
| 124 | if should_offer_reflect |
| 125 | else "No implicit trigger signal — only explicit phrases should invoke" |
| 126 | ), |
| 127 | } |
| 128 | |
| 129 | |
| 130 | SAMPLE_CONVERSATION = """User: Help me implement a sorting function. |
| 131 | Assistant: Sure. What language and what type of data? |
| 132 | User: Python, sorting a list of dicts by 'name' key. |
| 133 | Assistant: Here's the basic syntax: `sorted(items, key=lambda x: x['name'])` |
| 134 | User: That doesn't work — getting a TypeError. |
| 135 | Assistant: Can you share the exact error? Maybe some items don't have 'name'. |
| 136 | User: Ugh, still broken. Same error. |
| 137 | Assistant: Try adding a default: `key=lambda x: x.get('name', '')`. |
| 138 | User: Nope, that didn't work either. Different error now: KeyError on 'name'. |
| 139 | Assistant: That's strange. Let me look at your input data. |
| 140 | User: Frustrated. Let's try something else. |
| 141 | Assistant: OK let me reconsider. Are you sure the data is a list of dicts? |
| 142 | User: I think so. It's coming from an API. |
| 143 | Assistant: Let me check the API response shape. Add `print(type(items[0]))` first. |
| 144 | User: It's a string, not a dict. |
| 145 | Assistant: That's the issue. The API returns JSON strings; you need to parse them first. |
| 146 | User: This is weird, why isn't this working from the start? |
| 147 | Assistant: Let's go back to the data shape question first. |
| 148 | User: Going in circles. I'm stuck. |
| 149 | """ |
| 150 | |
| 151 | |
| 152 | def render_human(result: Dict[str, Any]) -> str: |
| 153 | out: List[str] = [] |
| 154 | t = result["turns"] |
| 155 | out.append(f"Conversation depth analysis:") |
| 156 | out.append(f" Total turns: {t['total_turns']} (user: {t['user_turns']}, assistant: {t['assistant_turns']})") |
| 157 | out.append(f" Max consecutive detail turns: {result['max_consecutive_detail_turns']}") |
| 158 | out.append(f" Frustration markers: {result['frustration_hits']}") |
| 159 | out.append(f" Dead-end / pivot markers: {result['dead_end_hits']}") |
| 160 | out.append("") |
| 161 | out.append(f"Implicit-trigger verdict: {result['implicit_trigger_verdict']}") |
| 162 | if result["trigger_signals"]: |
| 163 | out.append("Signals detected:") |
| 164 | for s in result["trigger_signals"]: |
| 165 | out.append(f" - {s}") |
| 166 | out.append("") |
| 167 | out.append(result["note"]) |
| 168 | return "\n".join(out) |
| 169 | |
| 170 | |
| 171 | def main(argv: List[str]) -> int: |
| 172 | parser = argparse.ArgumentParser(description=__doc__.split("\n")[0]) |
| 173 | parser.add_argument("--conversation", help="Path to conversation transcript text file") |
| 174 | parser.add_argument("--sample", action="store_true", help="Run on embedded sample (stuck-debugging scenario)") |
| 175 | parser.add_argument("--output", choices=["human", "json"], default="human") |
| 176 | args = parser.parse_args(argv) |
| 177 | |
| 178 | if args.sample: |
| 179 | text = SAMPLE_CONVERSATION |
| 180 | elif args.conversation: |
| 181 | p = Path(args.conversation) |
| 182 | if not p.exists(): |
| 183 | print(f"error: {args.conversation} not found", file=sys.stderr) |
| 184 | return 2 |
| 185 | text = p.read_text(encoding="utf-8") |
| 186 | else: |
| 187 | parser.print_help() |
| 188 | return 0 |
| 189 | |
| 190 | result = analyze(text) |
| 191 | if args.output == "json": |
| 192 | print(json.dumps(result, indent=2)) |
| 193 | else: |
| 194 | print(render_human(result)) |
| 195 | return 0 |
| 196 | |
| 197 | |
| 198 | if __name__ == "__main__": |
| 199 | sys.exit(main(sys.argv[1:])) |
文件内容是平台检查时读出的文本，只供查看。平台不会运行包里的任何脚本或程序。
