diff --git a/src/timmy/session_logger.py b/src/timmy/session_logger.py index efed01b..374a023 100644 --- a/src/timmy/session_logger.py +++ b/src/timmy/session_logger.py @@ -323,6 +323,75 @@ def session_history(query: str, role: str = "", limit: int = 10) -> str: _LOW_CONFIDENCE_THRESHOLD = 0.5 +def _categorize_entries( + entries: list[dict], +) -> tuple[list[dict], list[dict], list[dict], list[dict]]: + """Split session entries into messages, errors, timmy msgs, user msgs.""" + messages = [e for e in entries if e.get("type") == "message"] + errors = [e for e in entries if e.get("type") == "error"] + timmy_msgs = [e for e in messages if e.get("role") == "timmy"] + user_msgs = [e for e in messages if e.get("role") == "user"] + return messages, errors, timmy_msgs, user_msgs + + +def _find_low_confidence(timmy_msgs: list[dict]) -> list[dict]: + """Return Timmy responses below the confidence threshold.""" + return [ + m + for m in timmy_msgs + if m.get("confidence") is not None and m["confidence"] < _LOW_CONFIDENCE_THRESHOLD + ] + + +def _find_repeated_topics(user_msgs: list[dict], top_n: int = 5) -> list[tuple[str, int]]: + """Identify frequently mentioned words in user messages.""" + topic_counts: dict[str, int] = {} + for m in user_msgs: + for word in (m.get("content") or "").lower().split(): + cleaned = word.strip(".,!?\"'()[]") + if len(cleaned) > 3: + topic_counts[cleaned] = topic_counts.get(cleaned, 0) + 1 + return sorted( + ((w, c) for w, c in topic_counts.items() if c >= 3), + key=lambda x: x[1], + reverse=True, + )[:top_n] + + +def _format_reflection_section( + title: str, + items: list[dict], + formatter: object, + empty_msg: str, +) -> list[str]: + """Format a titled section with items, or an empty-state message.""" + if items: + lines = [f"### {title} ({len(items)})"] + for item in items[:5]: + lines.append(formatter(item)) # type: ignore[operator] + lines.append("") + return lines + return [f"### {title}\n{empty_msg}\n"] + + +def _build_insights( + low_conf: list[dict], + errors: list[dict], + repeated: list[tuple[str, int]], +) -> list[str]: + """Generate actionable insight bullets from analysis results.""" + insights: list[str] = [] + if low_conf: + insights.append("Consider studying topics where confidence was low.") + if errors: + insights.append("Review error patterns for recurring infrastructure issues.") + if repeated: + insights.append( + f'User frequently asks about "{repeated[0][0]}" — consider deepening knowledge here.' + ) + return insights or ["Conversations look healthy. Keep up the good work."] + + def self_reflect(limit: int = 30) -> str: """Review recent conversations and reflect on Timmy's own behavior. @@ -343,35 +412,12 @@ def self_reflect(limit: int = 30) -> str: if not entries: return "No conversation history to reflect on yet." - # Categorize entries - messages = [e for e in entries if e.get("type") == "message"] - errors = [e for e in entries if e.get("type") == "error"] - timmy_msgs = [e for e in messages if e.get("role") == "timmy"] - user_msgs = [e for e in messages if e.get("role") == "user"] - - # 1. Low-confidence responses - low_conf = [ - m - for m in timmy_msgs - if m.get("confidence") is not None and m["confidence"] < _LOW_CONFIDENCE_THRESHOLD - ] - - # 2. Identify repeated user topics (simple word frequency) - topic_counts: dict[str, int] = {} - for m in user_msgs: - for word in (m.get("content") or "").lower().split(): - cleaned = word.strip(".,!?\"'()[]") - if len(cleaned) > 3: - topic_counts[cleaned] = topic_counts.get(cleaned, 0) + 1 - repeated = sorted( - ((w, c) for w, c in topic_counts.items() if c >= 3), - key=lambda x: x[1], - reverse=True, - )[:5] + _messages, errors, timmy_msgs, user_msgs = _categorize_entries(entries) + low_conf = _find_low_confidence(timmy_msgs) + repeated = _find_repeated_topics(user_msgs) # Build reflection report sections: list[str] = ["## Self-Reflection Report\n"] - sections.append( f"Reviewed {len(entries)} recent entries: " f"{len(user_msgs)} user messages, " @@ -379,32 +425,27 @@ def self_reflect(limit: int = 30) -> str: f"{len(errors)} errors.\n" ) - # Low confidence - if low_conf: - sections.append(f"### Low-Confidence Responses ({len(low_conf)})") - for m in low_conf[:5]: - ts = (m.get("timestamp") or "?")[:19] - conf = m.get("confidence", 0) - text = (m.get("content") or "")[:120] - sections.append(f"- [{ts}] confidence={conf:.0%}: {text}") - sections.append("") - else: - sections.append( - "### Low-Confidence Responses\nNone found — all responses above threshold.\n" + sections.extend( + _format_reflection_section( + "Low-Confidence Responses", + low_conf, + lambda m: ( + f"- [{(m.get('timestamp') or '?')[:19]}] " + f"confidence={m.get('confidence', 0):.0%}: " + f"{(m.get('content') or '')[:120]}" + ), + "None found — all responses above threshold.", ) + ) + sections.extend( + _format_reflection_section( + "Errors", + errors, + lambda e: f"- [{(e.get('timestamp') or '?')[:19]}] {(e.get('error') or '')[:120]}", + "No errors recorded.", + ) + ) - # Errors - if errors: - sections.append(f"### Errors ({len(errors)})") - for e in errors[:5]: - ts = (e.get("timestamp") or "?")[:19] - err = (e.get("error") or "")[:120] - sections.append(f"- [{ts}] {err}") - sections.append("") - else: - sections.append("### Errors\nNo errors recorded.\n") - - # Repeated topics if repeated: sections.append("### Recurring Topics") for word, count in repeated: @@ -413,22 +454,8 @@ def self_reflect(limit: int = 30) -> str: else: sections.append("### Recurring Topics\nNo strong patterns detected.\n") - # Actionable summary - insights: list[str] = [] - if low_conf: - insights.append("Consider studying topics where confidence was low.") - if errors: - insights.append("Review error patterns for recurring infrastructure issues.") - if repeated: - top_topic = repeated[0][0] - insights.append( - f'User frequently asks about "{top_topic}" — consider deepening knowledge here.' - ) - if not insights: - insights.append("Conversations look healthy. Keep up the good work.") - sections.append("### Insights") - for insight in insights: + for insight in _build_insights(low_conf, errors, repeated): sections.append(f"- {insight}") return "\n".join(sections)