feat: add AI reasoning summary generator with fallback

- Implement ReasoningSummarizer class for generating 2-3 sentence AI summaries
- Add fallback to statistical summary when AI generation fails
- Format reasoning logs for summary prompt with truncation
- Handle empty reasoning logs with default message
- Add comprehensive unit tests with async mocking
This commit is contained in:
2025-11-03 23:16:15 -05:00
parent 5c19410f71
commit 197d3b7bf9
2 changed files with 190 additions and 0 deletions

View File

@@ -0,0 +1,110 @@
"""AI reasoning summary generation."""
import logging
from typing import List, Dict, Any
logger = logging.getLogger(__name__)
class ReasoningSummarizer:
"""Generate summaries of AI trading session reasoning."""
def __init__(self, model: Any):
"""Initialize summarizer.
Args:
model: LangChain chat model for generating summaries
"""
self.model = model
async def generate_summary(self, reasoning_log: List[Dict]) -> str:
"""Generate AI summary of trading session reasoning.
Args:
reasoning_log: List of message dicts with role and content
Returns:
Summary string (2-3 sentences)
"""
if not reasoning_log:
return "No trading activity recorded."
try:
# Build condensed version of reasoning log
log_text = self._format_reasoning_for_summary(reasoning_log)
summary_prompt = f"""You are reviewing your own trading decisions for the day.
Summarize your trading strategy and key decisions in 2-3 sentences.
Focus on:
- What you analyzed
- Why you made the trades you did
- Your overall strategy for the day
Trading session log:
{log_text}
Provide a concise summary:"""
response = await self.model.ainvoke([
{"role": "user", "content": summary_prompt}
])
# Extract content from response
if hasattr(response, 'content'):
return response.content
else:
return str(response)
except Exception as e:
logger.error(f"Failed to generate AI reasoning summary: {e}")
return self._generate_fallback_summary(reasoning_log)
def _format_reasoning_for_summary(self, reasoning_log: List[Dict]) -> str:
"""Format reasoning log into concise text for summary prompt.
Args:
reasoning_log: List of message dicts
Returns:
Formatted text representation
"""
formatted_parts = []
for msg in reasoning_log:
role = msg.get("role", "")
content = msg.get("content", "")
if role == "assistant":
# AI's thoughts
formatted_parts.append(f"AI: {content[:200]}")
elif role == "tool":
# Tool results
tool_name = msg.get("name", "tool")
formatted_parts.append(f"{tool_name}: {content[:100]}")
return "\n".join(formatted_parts)
def _generate_fallback_summary(self, reasoning_log: List[Dict]) -> str:
"""Generate simple statistical summary without AI.
Args:
reasoning_log: List of message dicts
Returns:
Fallback summary string
"""
trade_count = sum(
1 for msg in reasoning_log
if msg.get("role") == "tool" and msg.get("name") == "trade"
)
search_count = sum(
1 for msg in reasoning_log
if msg.get("role") == "tool" and msg.get("name") == "search"
)
return (
f"Executed {trade_count} trades using {search_count} market searches. "
f"Full reasoning log available."
)

View File

@@ -0,0 +1,80 @@
import pytest
from unittest.mock import AsyncMock, Mock
from agent.reasoning_summarizer import ReasoningSummarizer
class TestReasoningSummarizer:
@pytest.mark.asyncio
async def test_generate_summary_success(self):
"""Test successful AI summary generation."""
# Mock AI model
mock_model = AsyncMock()
mock_model.ainvoke.return_value = Mock(
content="Analyzed AAPL earnings. Bought 10 shares based on positive guidance."
)
summarizer = ReasoningSummarizer(model=mock_model)
reasoning_log = [
{"role": "user", "content": "Analyze market"},
{"role": "assistant", "content": "Let me check AAPL"},
{"role": "tool", "name": "search", "content": "AAPL earnings positive"}
]
summary = await summarizer.generate_summary(reasoning_log)
assert summary == "Analyzed AAPL earnings. Bought 10 shares based on positive guidance."
mock_model.ainvoke.assert_called_once()
@pytest.mark.asyncio
async def test_generate_summary_failure_fallback(self):
"""Test fallback summary when AI generation fails."""
# Mock AI model that raises exception
mock_model = AsyncMock()
mock_model.ainvoke.side_effect = Exception("API error")
summarizer = ReasoningSummarizer(model=mock_model)
reasoning_log = [
{"role": "assistant", "content": "Let me search"},
{"role": "tool", "name": "search", "content": "Results"},
{"role": "tool", "name": "trade", "content": "Buy AAPL"},
{"role": "tool", "name": "trade", "content": "Sell MSFT"}
]
summary = await summarizer.generate_summary(reasoning_log)
# Should return fallback with stats
assert "2 trades" in summary
assert "1 market searches" in summary
@pytest.mark.asyncio
async def test_format_reasoning_for_summary(self):
"""Test condensing reasoning log for summary prompt."""
mock_model = AsyncMock()
summarizer = ReasoningSummarizer(model=mock_model)
reasoning_log = [
{"role": "user", "content": "System prompt here"},
{"role": "assistant", "content": "I will analyze AAPL"},
{"role": "tool", "name": "search", "content": "AAPL earnings data..."},
{"role": "assistant", "content": "Based on analysis, buying AAPL"}
]
formatted = summarizer._format_reasoning_for_summary(reasoning_log)
# Should include key messages
assert "analyze AAPL" in formatted
assert "search" in formatted
assert "buying AAPL" in formatted
@pytest.mark.asyncio
async def test_empty_reasoning_log(self):
"""Test handling empty reasoning log."""
mock_model = AsyncMock()
summarizer = ReasoningSummarizer(model=mock_model)
summary = await summarizer.generate_summary([])
assert summary == "No trading activity recorded."