reasoning-trace-optimizer
by Murat Can Koylan
Debug and optimize AI agents by analyzing reasoning traces, context degradation, tool confusion, instruction drift, repeated task failures, and performance regressions.
Reaches the networkRuns commands13 scripts4 findings
Labels from its files: reaches the network, runs commands. 63 files, read Oct 21, 2026. Nothing was run.
npx skills add https://github.com/muratcankoylan/Agent-Skills-for-Context-Engineering --skill interleaved-thinkingWhat it lets an agent do
Each label, and the line in the skill's files that set it.
- Reaches the network
Its scripts or instructions fetch from or send to the network.
- examples/01_basic_capture.py makes network calls.
- examples/02_tool_usage.py makes network calls.
- examples/03_full_optimization.py makes network calls.
- reasoning_trace_optimizer/analyzer.py makes network calls.
- reasoning_trace_optimizer/capture.py makes network calls.
- reasoning_trace_optimizer/cli.py makes network calls.
- Runs commands
Allows a shell or ships scripts the agent runs.
- It ships 13 scripts the agent can run.
Findings (4)
Matches of fixed checks, with the text they matched.
- highexamples/01_basic_capture.py
examples/01_basic_capture.py mentions a secret on disk and also makes network calls.
…ath(__file__).parent.parent / ".env" load_dotenv(env_path) def …
What this check looks for - highexamples/02_tool_usage.py
examples/02_tool_usage.py mentions a secret on disk and also makes network calls.
…ath(__file__).parent.parent / ".env" load_dotenv(env_path) # De…
What this check looks for - warnexamples/03_full_optimization.py
examples/03_full_optimization.py runs a command string through a shell, or evaluates code it builds at run time.
… } result = eval(expression, {"__builtins__": {…
What this check looks for - highexamples/03_full_optimization.py
examples/03_full_optimization.py mentions a secret on disk and also makes network calls.
…ath(__file__).parent.parent / ".env" load_dotenv(env_path) # ==…
What this check looks for
Not a safety rating, and no finding is not a safety guarantee. These are fixed checks of the files as read on Oct 21, 2026. Nothing was run. How skills are checked.
SKILL.md
As the publisher wrote it. Shown as text, never rendered.
---
name: reasoning-trace-optimizer
description: "Debug and optimize AI agents by analyzing reasoning traces, context degradation, tool confusion, instruction drift, repeated task failures, and performance regressions."
---
# Reasoning Trace Optimizer
Debug and optimize AI agents by analyzing their reasoning traces. This skill uses MiniMax M2.1's interleaved thinking to provide deep insight into agent decision-making and generate concrete improvements.
## When to Activate
- Agent reasoning traces need debugging, analysis, or prompt optimization
- Agent task fails and user wants to understand why
- User mentions "context degradation", "tool confusion", or "instruction drift"
- Request to improve agent performance or reduce errors
- User wants to generate shareable learnings from debugging sessions
- After repeated failures on similar tasks
## Core Concepts
### Interleaved Thinking
Unlike standard reasoning models that think once at the start, interleaved thinking allows reasoning BETWEEN each tool interaction. This is critical because:
1. **Long-horizon tasks** require maintaining focus across many turns
2. **External perturbations** (tool outputs, environment changes) need real-time adaptation
3. **Debugging** requires seeing HOW decisions were made, not just WHAT was output
### The Optimization Loop
```
Execute Agent → Capture Traces → Analyze Patterns → Optimize Prompt → Re-run
↑____________|
```
Each iteration improves the prompt based on detected patterns until convergence.
### Pattern Detection
Common failure patterns the analyzer detects:
| Pattern | Description |
|---------|-------------|
| `context_degradation` | Model loses track of information over long contexts |
| `tool_confusion` | Model misunderstands tool capabilities or outputs |
| `instruction_drift` | Model gradually deviates from original instructions |
| `goal_abandonment` | Model stops pursuing the original goal |
| `circular_reasoning` | Model repeats similar actions without progress |
| `premature_conclusion` | Model concludes before completing the task |
## Usage Modes
### Mode 1: M2.1 Agent Debugging
Run a task through M2.1 and analyze its reasoning:
```python
from reasoning_trace_optimizer import TraceCapture, TraceAnalyzer
capture = TraceCapture()
trace = capture.run(
task="Search for Python tutorials and summarize them",
system_prompt="You are a research assistant.",
tools=[search_tool],
tool_executor=execute_search
)
analyzer = TraceAnalyzer()
analysis = analyzer.analyze(trace)
print(f"Score: {analysis.overall_score}/100")
for pattern in analysis.patterns:
print(f"Found: {pattern.type.value} - {pattern.suggestion}")
```
### Mode 2: Full Optimization Loop
Automatically iterate until the prompt is optimized:
```python
from reasoning_trace_optimizer import OptimizationLoop, LoopConfig
config = LoopConfig(
max_iterations=5,
min_score_threshold=80.0,
)
loop = OptimizationLoop(config=config)
result = loop.run(
task="Analyze this codebase and suggest improvements",
initial_prompt="You are a code reviewer.",
tools=[read_file_tool, search_tool],
tool_executor=execute_tool
)
print(f"Improved: {result.initial_score} → {result.final_score}")
print(f"Final prompt:\n{result.final_prompt}")
```
### Mode 3: Universal Session Analysis
Analyze any agent's previous thinking (works with Claude, GPT, etc.):
When this skill is activated in Claude Code, it can analyze the current session's thinking blocks to identify issues and suggest improvements.
```
/reasoning-trace-optimizer analyze-session
```
### Mode 4: Generate Shareable Skills
Convert optimization learnings into reusable Agent Skills:
```python
from reasoning_trace_optimizer import SkillGenerator
generator = SkillGenerator()
skill_path = generator.generate(
result=loop_result,
skill_name="web-search-best-practices",
output_dir="./skills"
)
```
## CLI Commands
```bash
# Capture reasoning trace
rto capture "Search for Python tutorials" -s "You are a helpful assistant."
# Analyze a task
rto analyze "Debug this code" -o analysis.txt
# Run optimization loop
rto optimize "Research AI papers" --max-iterations 5 --generate-skill
# Generate skill from artifacts
rto generate-skill my-skill-name --artifacts-dir ./optimization_artifacts
```
## Integration with Claude Code
### Auto-trigger on Failure
Add to your hooks to automatically analyze failures:
```json
{
"hooks": {
"post_tool_error": {
"command": "rto analyze-session --last-error"
}
}
}
```
### On-demand Analysis
Use the slash command to analyze current session:
```
/reasoning-trace-optimizer
```
This will:
1. Extract thinking blocks from the current session
2. Identify patterns and issues
3. Suggest prompt improvements
4. Optionally update the system prompt
## Guidelines
1. **Preserve full context**: M2.1 requires full response history including thinking blocks for optimal performance
2. **Use appropriate tools**: Define tools clearly with unambiguous descriptions
3. **Set realistic convergence thresholds**: 5-10% improvement per iteration is typical
4. **Review generated skills**: Auto-generated skills should be reviewed before sharing
5. **Monitor token usage**: Each optimization iteration uses significant tokens
## Examples
### Before Optimization
```
System: You are a helpful assistant.
Issue: Agent called wrong tools, lost track of goal after 3 turns
Score: 45/100
Patterns: tool_confusion, goal_abandonment
```
### After Optimization
```
System: You are a research assistant focused on finding accurate information.
IMPORTANT GUIDELINES:
- Always verify search results before summarizing
- If a tool returns an error, try an alternative approach
- Keep track of your original goal throughout the task
- Validate findings against multiple sources when possible
Issue: None
Score: 85/100
Patterns: None detected
```
## References
- MiniMax M2.1 Documentation: https://platform.minimax.io/docs
- Interleaved Thinking Guide: See `docs/interleavedthinking.md`
- Agent Generalization: See `docs/agentthinking.md`
---
## Skill Metadata
**Created**: 2025-01-11
**Author**: Muratcan Koylan
**Version**: 0.1.0
**Powered by**: MiniMax M2.1
**Partnership**: Built in collaboration with MiniMax AI
Scripts (12)
Read as text. anlyon.dev never runs them.
examples/01_basic_capture.pyShowHide
"""
Example 1: Basic Trace Capture
Demonstrates capturing reasoning traces from M2.1 for a simple task.
This shows how interleaved thinking provides visibility into agent decisions.
"""
import os
from pathlib import Path
from dotenv import load_dotenv
from reasoning_trace_optimizer import TraceCapture
from reasoning_trace_optimizer.capture import format_trace_for_display
# Load environment variables from the project root
env_path = Path(__file__).parent.parent / ".env"
load_dotenv(env_path)
def main():
"""Run a simple task and capture the reasoning trace."""
# Initialize capture with M2.1
capture = TraceCapture(
api_key=os.getenv("ANTHROPIC_API_KEY"),
base_url="https://api.minimax.io/anthropic",
model="MiniMax-M2.1",
)
# Define a simple task
task = "Explain what interleaved thinking is and why it matters for AI agents."
system_prompt = "You are an AI researcher explaining concepts clearly."
print("=" * 60)
print("BASIC TRACE CAPTURE EXAMPLE")
print("=" * 60)
print(f"\nTask: {task}")
print(f"System Prompt: {system_prompt}")
print("\nCapturing reasoning trace...\n")
# Capture the trace
trace = capture.run(
task=task,
system_prompt=system_prompt,
max_turns=5,
)
# Display the trace
print(format_trace_for_display(trace))
# Summary statistics
print("\n" + "=" * 60)
print("TRACE STATISTICS")
print("=" * 60)
print(f"Session ID: {trace.session_id}")
print(f"Model: {trace.model}")
print(f"Success: {trace.success}")
print(f"Total Turns: {trace.total_turns}")
print(f"Thinking Blocks: {len(trace.thinking_blocks)}")
print(f"Tool Calls: {len(trace.tool_calls)}")
print(f"Total Tokens: {trace.total_tokens}")
# Show each thinking block summary
if trace.thinking_blocks:
print("\n" + "=" * 60)
print("THINKING BLOCK SUMMARIES")
print("=" * 60)
for i, thinking in enumerate(trace.thinking_blocks):
print(f"\n[Turn {thinking.turn_index}] ({len(thinking.content)} chars)")
# Show first 200 chars
preview = thinking.content[:200].replace("\n", " ")
print(f" Preview: {preview}...")
if __name__ == "__main__":
main()
examples/02_tool_usage.pyShowHide
"""
Example 2: Tool Usage with Trace Capture
Demonstrates how M2.1's interleaved thinking reasons between tool calls.
This is where interleaved thinking really shines - you can see the model
adapting to tool outputs in real-time.
"""
import json
import os
from pathlib import Path
from dotenv import load_dotenv
from reasoning_trace_optimizer import TraceCapture
from reasoning_trace_optimizer.capture import format_trace_for_display
# Load environment variables from the project root
env_path = Path(__file__).parent.parent / ".env"
load_dotenv(env_path)
# Define mock tools
TOOLS = [
{
"name": "get_weather",
"description": "Get current weather for a location. Returns temperature and conditions.",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City name, e.g., 'San Francisco, CA'",
}
},
"required": ["location"],
},
},
{
"name": "get_forecast",
"description": "Get 3-day weather forecast for a location.",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City name",
},
"days": {
"type": "integer",
"description": "Number of days (1-3)",
"default": 3,
},
},
"required": ["location"],
},
},
]
# Mock tool executor
def execute_tool(name: str, input_data: dict) -> str:
"""Execute a mock tool and return results."""
if name == "get_weather":
location = input_data.get("location", "Unknown")
# Simulate different weather for different cities
if "san francisco" in location.lower():
return json.dumps({
"location": location,
"temperature": "18°C",
"conditions": "Foggy",
"humidity": "85%",
})
elif "new york" in location.lower():
return json.dumps({
"location": location,
"temperature": "22°C",
"conditions": "Partly cloudy",
"humidity": "60%",
})
else:
return json.dumps({
"location": location,
"temperature": "20°C",
"conditions": "Clear",
"humidity": "50%",
})
elif name == "get_forecast":
location = input_data.get("location", "Unknown")
days = input_data.get("days", 3)
forecast = []
for i in range(days):
forecast.append({
"day": i + 1,
"high": f"{20 + i * 2}°C",
"low": f"{12 + i}°C",
"conditions": ["Sunny", "Cloudy", "Rainy"][i % 3],
})
return json.dumps({
"location": location,
"forecast": forecast,
})
return json.dumps({"error": f"Unknown tool: {name}"})
def main():
"""Run a task with tools and observe interleaved thinking."""
capture = TraceCapture(
api_key=os.getenv("ANTHROPIC_API_KEY"),
base_url="https://api.minimax.io/anthropic",
model="MiniMax-M2.1",
)
task = """Compare the current weather in San Francisco and New York City.
Then tell me which city would be better for outdoor activities this weekend."""
system_prompt = """You are a helpful weather assistant.
Use the available tools to get accurate weather information.
Always provide specific data to support your recommendations."""
print("=" * 60)
print("TOOL USAGE WITH INTERLEAVED THINKING")
print("=" * 60)
print(f"\nTask: {task}")
print(f"\nTools available: {', '.join(t['name'] for t in TOOLS)}")
print("\nCapturing trace with tool usage...\n")
# Capture the trace (using non-streaming for reliability)
trace = capture.run(
task=task,
system_prompt=system_prompt,
tools=TOOLS,
tool_executor=execute_tool,
max_turns=10,
)
print("\n\n" + "=" * 60)
print("TRACE ANALYSIS")
print("=" * 60)
print(f"\nSuccess: {trace.success}")
print(f"Total Turns: {trace.total_turns}")
print(f"Thinking Blocks: {len(trace.thinking_blocks)}")
print(f"Tool Calls: {len(trace.tool_calls)}")
# Show how thinking evolved between tool calls
print("\n" + "=" * 60)
print("THINKING EVOLUTION ACROSS TOOL CALLS")
print("=" * 60)
for i, thinking in enumerate(trace.thinking_blocks):
print(f"\n[Turn {thinking.turn_index}] Thinking Block {i + 1}")
print("-" * 40)
# Show what tool was called after this thinking
turn_tools = trace.get_tool_calls_at_turn(thinking.turn_index)
if turn_tools:
print(f"Following action: Called {', '.join(t.name for t in turn_tools)}")
else:
print("Following action: Generated response")
# Show key reasoning points (first 300 chars)
print(f"\nReasoning preview:\n{thinking.content[:300]}...")
# Show tool call results
print("\n" + "=" * 60)
print("TOOL CALL SUMMARY")
print("=" * 60)
for tc in trace.tool_calls:
status = "✅" if tc.success else "❌"
print(f"\n{status} {tc.name}")
print(f" Input: {json.dumps(tc.input)}")
print(f" Result: {tc.result[:100]}..." if tc.result and len(tc.result) > 100 else f" Result: {tc.result}")
# Final response
if trace.final_response:
print("\n" + "=" * 60)
print("FINAL RESPONSE")
print("=" * 60)
print(trace.final_response)
if __name__ == "__main__":
main()
examples/03_full_optimization.pyShowHide
"""
Example 3: Full Optimization Loop with Comprehensive Tools
Demonstrates the complete optimization cycle with realistic tools:
- Web search for finding information
- URL reading for fetching content
- File system operations (read, write, list)
- Note-taking for tracking findings
This example uses REAL URLs and realistic content to demonstrate
how the Reasoning Trace Optimizer works in production scenarios.
"""
import json
import os
import random
from datetime import datetime
from pathlib import Path
from dotenv import load_dotenv
from reasoning_trace_optimizer import (
OptimizationLoop,
LoopConfig,
SkillGenerator,
)
# Load environment variables from the project root
env_path = Path(__file__).parent.parent / ".env"
load_dotenv(env_path)
# =============================================================================
# COMPREHENSIVE TOOL DEFINITIONS
# =============================================================================
TOOLS = [
# Web Search Tool
{
"name": "web_search",
"description": "Search the web for information. Returns a list of results with titles, URLs, and snippets. Use specific queries for better results.",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query - be specific and use relevant keywords",
},
"num_results": {
"type": "integer",
"description": "Number of results to return (1-10, default 5)",
"default": 5,
},
},
"required": ["query"],
},
},
# Read URL Tool
{
"name": "read_url",
"description": "Fetch and read the content of a webpage. Returns the main text content. Use after web_search to get full details from a result.",
"input_schema": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "The URL to fetch content from",
},
},
"required": ["url"],
},
},
# File Read Tool
{
"name": "read_file",
"description": "Read the contents of a local file. Supports text files, markdown, JSON, etc.",
"input_schema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the file to read",
},
},
"required": ["path"],
},
},
# File Write Tool
{
"name": "write_file",
"description": "Write content to a local file. Creates the file if it doesn't exist, overwrites if it does.",
"input_schema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path where to write the file",
},
"content": {
"type": "string",
"description": "Content to write to the file",
},
},
"required": ["path", "content"],
},
},
# List Directory Tool
{
"name": "list_directory",
"description": "List files and folders in a directory. Useful for exploring project structure.",
"input_schema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Directory path to list (default: current directory)",
"default": ".",
},
},
"required": [],
},
},
# Save Note Tool
{
"name": "save_note",
"description": "Save a research note with title and content. Use to track important findings during research.",
"input_schema": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "Title of the note",
},
"content": {
"type": "string",
"description": "Content of the note",
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "Optional tags for categorization",
},
},
"required": ["title", "content"],
},
},
# Calculator Tool
{
"name": "calculator",
"description": "Perform mathematical calculations. Supports basic arithmetic and common functions.",
"input_schema": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Mathematical expression to evaluate (e.g., '2 + 2', 'sqrt(16)', '100 * 0.15')",
},
},
"required": ["expression"],
},
},
]
# =============================================================================
# REAL-WORLD SIMULATED DATA
# Based on actual documentation and research from AI companies
# =============================================================================
# Simulated web search results with REAL URLs
SEARCH_DATABASE = {
"context engineering ai": [
{
"title": "Context Engineering for AI Agents - Anthropic",
"url": "https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching",
"snippet": "Prompt caching is a feature that optimizes API usage by allowing resuming from specific prefixes in your prompts. Cache the context you want to reuse across requests.",
},
{
"title": "Building Effective AI Agents - Anthropic Research",
"url": "https://www.anthropic.com/research/building-effective-agents",
"snippet": "A comprehensive guide to building effective AI agents. Covers tool use, context management, error handling, and best practices for production deployments.",
},
{
"title": "Large Language Models and Context Windows - OpenAI",
"url": "https://platform.openai.com/docs/guides/text-generation",
"snippet": "Understanding how context windows work in large language models. Learn about token limits, context management strategies, and optimizing for performance.",
},
],
"interleaved thinking agents": [
{
"title": "MiniMax M2.1 - Interleaved Thinking Model",
"url": "https://www.minimax.io/platform/docs/M2.1",
"snippet": "M2.1 introduces interleaved thinking - the ability for models to reason between tool calls, enabling better debugging and adaptability in agentic workflows.",
},
{
"title": "Chain of Thought Prompting - Google Research",
"url": "https://arxiv.org/abs/2201.11903",
"snippet": "Chain-of-thought prompting enables complex reasoning in large language models. This paper explores how step-by-step reasoning improves model performance.",
},
],
"prompt optimization techniques": [
{
"title": "Prompt Engineering Guide - DAIR.AI",
"url": "https://www.promptingguide.ai/techniques",
"snippet": "Comprehensive guide to prompt engineering techniques including zero-shot, few-shot, chain-of-thought, and advanced methods for optimizing LLM outputs.",
},
{
"title": "Best Practices for Prompt Engineering - OpenAI",
"url": "https://platform.openai.com/docs/guides/prompt-engineering",
"snippet": "Official OpenAI guide on prompt engineering best practices. Covers strategies for getting better results, handling edge cases, and iterative refinement.",
},
],
"agent debugging best practices": [
{
"title": "Debugging AI Agents - LangChain Documentation",
"url": "https://python.langchain.com/docs/how_to/debugging",
"snippet": "Learn how to debug LangChain agents effectively. Covers tracing, verbose mode, callbacks, and common debugging patterns for complex agent workflows.",
},
{
"title": "LLM Observability and Tracing - Weights & Biases",
"url": "https://docs.wandb.ai/guides/prompts",
"snippet": "Track and debug LLM applications with W&B Prompts. Visualize chains, compare outputs, and identify failure patterns in your AI applications.",
},
],
"context window optimization": [
{
"title": "Claude's Context Window - Anthropic Documentation",
"url": "https://docs.anthropic.com/en/docs/build-with-claude/context-windows",
"snippet": "Claude supports context windows up to 200K tokens. Learn how to effectively use large context windows and optimize token usage for cost and performance.",
},
{
"title": "Lost in the Middle: How Language Models Use Long Contexts",
"url": "https://arxiv.org/abs/2307.03172",
"snippet": "Research on how LLMs utilize information across long contexts. Models perform worse when relevant info is in the middle vs. beginning/end of context.",
},
],
}
# Simulated webpage content based on REAL documentation
PAGE_CONTENT = {
"https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching": """
# Prompt Caching - Anthropic Documentation
Prompt caching is a feature that optimizes API usage by allowing you to cache frequently used context.
## Overview
Prompt caching allows you to cache the system prompt, examples, and other static content that remains constant across multiple requests. This:
- **Reduces latency** by up to 85% for cached content
- **Lowers costs** by avoiding re-processing of identical context
- **Improves throughput** for high-volume applications
## How It Works
When you enable prompt caching, the API stores a hash of your prompt prefix. On subsequent requests with the same prefix, the cached computation is reused.
### Cache Breakpoints
You can specify cache breakpoints using the `cache_control` parameter:
```python
messages = [
{
"role": "user",
"content": [
{
"type": "text",
"text": "Your static context here...",
"cache_control": {"type": "ephemeral"}
}
]
}
]
```
## Best Practices
1. **Cache stable content**: Put instructions and examples that don't change in the cached portion
2. **Place dynamic content last**: User queries and variable data should come after cached content
3. **Monitor cache hits**: Use the response headers to track cache efficiency
4. **Minimum cache size**: Content must be at least 1024 tokens to be cached
## Context Engineering Implications
Effective prompt caching is a key part of context engineering. By understanding what to cache:
- System prompts with role definitions
- Tool descriptions that remain constant
- Few-shot examples for consistent behavior
- Reference documentation the model needs
You reduce both latency and cost while maintaining quality.
""",
"https://www.anthropic.com/research/building-effective-agents": """
# Building Effective AI Agents - Anthropic Research
This guide covers best practices for building reliable, effective AI agents using Claude.
## Core Principles
### 1. Start Simple, Add Complexity Gradually
Begin with the simplest possible agent architecture:
- Single tool with clear purpose
- Linear workflow without branching
- Explicit success criteria
Only add complexity when you have evidence it's needed.
### 2. Tool Design Matters
Well-designed tools make agents more reliable:
- **Clear descriptions**: Explain what the tool does AND when to use it
- **Typed inputs**: Use JSON Schema to define expected parameters
- **Informative outputs**: Return data the model can interpret and act on
- **Error messages**: Provide actionable guidance when things fail
### 3. Context Management
Context is your most precious resource:
- **Token efficiency**: Every token costs money and attention
- **Structured format**: Use consistent formatting for easier parsing
- **Progressive disclosure**: Load information on-demand
- **Summarization**: Compress long histories while preserving key facts
### 4. Error Handling
Agents will encounter errors. Design for recovery:
- Give the model explicit permission to retry
- Provide diagnostic information in error messages
- Set clear stopping conditions to prevent infinite loops
- Log everything for debugging
## Common Anti-Patterns
1. **Over-engineering**: Building complex multi-agent systems before validating single-agent performance
2. **Vague tools**: Tool descriptions that don't clarify when to use each tool
3. **Context overload**: Stuffing too much information into the prompt
4. **No exit conditions**: Letting agents run indefinitely without progress checks
## Debugging Strategies
### Trace Analysis
The key to debugging agents is understanding their reasoning:
1. Capture the full reasoning trace including thinking blocks
2. Identify where the agent's understanding diverged from reality
3. Look for patterns: tool confusion, goal drift, context loss
4. Iterate on prompts based on specific failure modes
### Interleaved Thinking
Models with interleaved thinking (reasoning between tool calls) provide better debugging insight because you can see:
- How they interpreted each tool result
- What alternatives they considered
- When and why they changed approach
""",
"https://platform.openai.com/docs/guides/text-generation": """
# Text Generation - OpenAI Documentation
Learn how to generate text with OpenAI's models.
## Context Windows
Each model has a context window that determines the maximum number of tokens it can process:
| Model | Context Window |
|-------|----------------|
| GPT-4o | 128K tokens |
| GPT-4 Turbo | 128K tokens |
| GPT-3.5 Turbo | 16K tokens |
### Managing Context
For long conversations or documents:
1. **Truncation**: Remove oldest messages when approaching the limit
2. **Summarization**: Replace old messages with summaries
3. **Retrieval**: Use RAG to fetch only relevant content
### Token Counting
Use the tiktoken library to count tokens before sending requests:
```python
import tiktoken
encoding = tiktoken.encoding_for_model("gpt-4")
num_tokens = len(encoding.encode("Your text here"))
```
## Best Practices
### Structured Prompts
Organize your prompts with clear sections:
- System message: Role and general instructions
- Context: Background information needed
- Task: Specific request with format requirements
- Examples: Few-shot demonstrations if helpful
### Temperature and Sampling
- **temperature=0**: Deterministic, best for factual tasks
- **temperature=0.7**: Balanced creativity and coherence
- **temperature=1.0+**: More random, for creative tasks
""",
"https://www.minimax.io/platform/docs/M2.1": """
# MiniMax M2.1 - Interleaved Thinking Model
M2.1 is a next-generation reasoning model that introduces **interleaved thinking** - continuous reasoning throughout task execution.
## What is Interleaved Thinking?
Traditional reasoning models think once at the start, then execute:
```
Think → Act → Act → Act → Done
```
M2.1 thinks between every action:
```
Think → Act → Think → Act → Think → Act → Done
```
## Why This Matters
### 1. Better Debugging
The thinking blocks expose the model's reasoning process. You can see:
- What it understood from tool results
- How it decided what to do next
- Where it might have gone wrong
### 2. Adaptive Behavior
By reasoning after each tool call, M2.1 can:
- React to unexpected outputs
- Recover from errors mid-execution
- Adjust strategy based on new information
### 3. Long-Horizon Tasks
For complex multi-step tasks, maintaining focus is crucial. Interleaved thinking:
- Reinforces the original goal
- Tracks progress toward completion
- Identifies when the task is done
## API Usage
### Anthropic SDK
```python
import anthropic
client = anthropic.Anthropic(
api_key="your-key",
base_url="https://api.minimax.io/anthropic"
)
response = client.messages.create(
model="MiniMax-M2.1",
max_tokens=4096,
messages=[{"role": "user", "content": "Your task"}]
)
# Access thinking blocks
for block in response.content:
if block.type == "thinking":
print(f"Thinking: {block.thinking}")
elif block.type == "text":
print(f"Response: {block.text}")
```
## Best Practices
1. **Preserve full context**: Always include thinking blocks in message history
2. **Clear tool descriptions**: Help the model understand when to use each tool
3. **Explicit success criteria**: Define what "done" looks like
4. **Error guidance**: Give clear instructions for handling failures
""",
"https://www.promptingguide.ai/techniques": """
# Prompt Engineering Techniques - DAIR.AI
A comprehensive guide to prompt engineering techniques for large language models.
## Basic Techniques
### Zero-Shot Prompting
Ask the model to perform a task without examples:
```
Classify this text as positive, negative, or neutral:
"I really enjoyed the movie but the ending was disappointing."
```
### Few-Shot Prompting
Provide examples to guide the model:
```
Classify sentiment:
"Great product!" → Positive
"Terrible service." → Negative
"It was okay." → Neutral
"I really enjoyed the movie but the ending was disappointing." →
```
## Advanced Techniques
### Chain-of-Thought (CoT)
Encourage step-by-step reasoning:
```
Solve this problem step by step:
If John has 5 apples and gives 2 to Mary, then buys 3 more, how many does he have?
Let's think through this:
1. John starts with 5 apples
2. He gives 2 to Mary: 5 - 2 = 3 apples
3. He buys 3 more: 3 + 3 = 6 apples
Answer: 6 apples
```
### Self-Consistency
Generate multiple reasoning paths and take the majority answer. Improves reliability for complex reasoning tasks.
### Tree of Thoughts
Explore multiple reasoning branches simultaneously, evaluating and pruning paths to find optimal solutions.
## Prompt Optimization
### Iterative Refinement
1. Start with a basic prompt
2. Test on representative examples
3. Analyze failures
4. Refine prompt based on patterns
5. Repeat until convergence
### Common Failure Patterns
| Pattern | Solution |
|---------|----------|
| Goal drift | Add explicit goal reminders |
| Hallucination | Require source citations |
| Incomplete output | Specify format requirements |
| Wrong tool usage | Improve tool descriptions |
""",
"https://platform.openai.com/docs/guides/prompt-engineering": """
# Prompt Engineering Best Practices - OpenAI
Official guide to getting better results from large language models.
## Six Strategies
### 1. Write Clear Instructions
Be specific about what you want:
- Include details in your query
- Ask the model to adopt a persona
- Use delimiters to mark distinct sections
- Specify desired output format and length
### 2. Provide Reference Text
Reduce hallucinations:
- Instruct the model to answer using provided text
- Ask for citations from the source material
- Use retrieval to inject relevant context
### 3. Split Complex Tasks
Break down hard problems:
- Use intent classification to route queries
- Summarize long documents in chunks
- Break multi-step tasks into sequential prompts
### 4. Give the Model Time to Think
Improve reasoning:
- Ask for a chain of reasoning
- Use inner monologue to hide intermediate steps
- Ask if previous steps were correct
### 5. Use External Tools
Augment model capabilities:
- Use code execution for accurate calculations
- Use retrieval for up-to-date information
- Use … [19936 more characters not shown]reasoning_trace_optimizer/__init__.pyShowHide
"""
Reasoning Trace Optimizer
Debug and optimize AI agents by analyzing reasoning traces
using MiniMax M2.1's interleaved thinking capabilities.
"""
from reasoning_trace_optimizer.models import (
ReasoningTrace,
ThinkingBlock,
ToolCall,
Pattern,
PatternType,
Severity,
AnalysisResult,
OptimizationResult,
PromptDiff,
LoopIteration,
LoopResult,
)
from reasoning_trace_optimizer.capture import TraceCapture
from reasoning_trace_optimizer.analyzer import TraceAnalyzer
from reasoning_trace_optimizer.optimizer import PromptOptimizer
from reasoning_trace_optimizer.loop import OptimizationLoop, LoopConfig
from reasoning_trace_optimizer.skill_generator import SkillGenerator
__version__ = "0.1.0"
__all__ = [
# Models
"ReasoningTrace",
"ThinkingBlock",
"ToolCall",
"Pattern",
"PatternType",
"Severity",
"AnalysisResult",
"OptimizationResult",
"PromptDiff",
"LoopIteration",
"LoopResult",
# Capture
"TraceCapture",
# Analyzer
"TraceAnalyzer",
# Optimizer
"PromptOptimizer",
# Loop
"OptimizationLoop",
"LoopConfig",
# Skill Generator
"SkillGenerator",
]
reasoning_trace_optimizer/analyzer.pyShowHide
"""
TraceAnalyzer: Analyzes reasoning traces to detect patterns and issues.
Uses M2.1's own interleaved thinking to analyze agent reasoning traces,
detecting patterns like context degradation, tool confusion, and instruction drift.
"""
import json
import os
from typing import Any
import anthropic
from reasoning_trace_optimizer.models import (
AnalysisResult,
Pattern,
PatternType,
ReasoningTrace,
Severity,
)
ANALYSIS_SYSTEM_PROMPT = """You are an expert AI agent debugger specializing in analyzing reasoning traces.
Your task is to analyze an agent's interleaved thinking trace and identify:
1. **Patterns of failure** - detect specific failure modes with evidence
2. **Quality scores** - rate the agent's reasoning on multiple dimensions
3. **Actionable recommendations** - specific improvements for prompts/instructions
## Pattern Definitions
Detect these patterns with specific evidence from thinking blocks:
- **context_degradation**: Agent loses or forgets information from earlier in the conversation
- Look for: Repeated questions, contradicting earlier statements, missing key details
- **tool_confusion**: Agent misunderstands what a tool does or how to use it
- Look for: Wrong tool selection, incorrect parameters, misinterpreting results
- **instruction_drift**: Agent gradually deviates from original instructions/persona
- Look for: Changing behavior, ignoring constraints, different tone over time
- **hallucination**: Agent generates information not supported by context or tools
- Look for: Made-up facts, fabricated tool results, unsourced claims
- **incomplete_reasoning**: Agent reaches conclusions without thorough analysis
- Look for: Skipped steps, missing validation, superficial exploration
- **tool_misuse**: Agent uses tools incorrectly or inefficiently
- Look for: Redundant calls, wrong parameters, unused results
- **goal_abandonment**: Agent stops pursuing the original objective
- Look for: Topic drift, giving up, switching goals without reason
- **circular_reasoning**: Agent repeats similar actions without progress
- Look for: Same queries repeated, looping behavior, no new information
- **premature_conclusion**: Agent concludes before completing the task
- Look for: Early stops, incomplete answers, skipped requirements
- **missing_validation**: Agent doesn't verify results or assumptions
- Look for: No cross-checking, accepting first result, no error handling
## Analysis Focus
You have access to the FULL reasoning trace including all thinking blocks between tool calls.
This gives you unique insight into HOW the agent reasons, not just what it outputs.
For each thinking block, examine:
- What is the agent's current understanding?
- How does it interpret tool results?
- What alternatives does it consider?
- Does it maintain awareness of the original goal?
Provide your analysis in the specified JSON format with concrete evidence."""
ANALYSIS_PROMPT_TEMPLATE = """Analyze the following agent reasoning trace:
## Task
{task}
## System Prompt Given to Agent
{system_prompt}
## Reasoning Trace
{trace}
## Tool Calls Made
{tool_calls}
## Final Outcome
Success: {success}
Final Response: {final_response}
Error (if any): {error}
---
Provide your analysis as JSON with this exact structure:
```json
{{
"patterns": [
{{
"type": "<one of: context_degradation, tool_confusion, instruction_drift, hallucination, incomplete_reasoning, tool_misuse, goal_abandonment, circular_reasoning, premature_conclusion, missing_validation>",
"severity": "<one of: low, medium, high, critical>",
"description": "<what the pattern is>",
"evidence": ["<excerpt from thinking>", "<another excerpt>"],
"turn_indices": [0, 2],
"suggestion": "<how to fix this>",
"confidence": 0.85
}}
],
"scores": {{
"reasoning_clarity": 75,
"goal_adherence": 80,
"tool_usage_quality": 60,
"error_recovery": 50,
"overall": 66
}},
"strengths": ["<strength 1>", "<strength 2>"],
"weaknesses": ["<weakness 1>", "<weakness 2>"],
"recommendations": [
"<specific actionable recommendation>",
"<another recommendation>"
]
}}
```
Think carefully about each aspect before providing your analysis."""
class TraceAnalyzer:
"""
Analyzes reasoning traces using M2.1 to detect patterns and score quality.
The analyzer uses M2.1's interleaved thinking to deeply understand
the agent's reasoning process and identify issues that wouldn't be
visible from outputs alone.
Example:
```python
analyzer = TraceAnalyzer()
result = analyzer.analyze(trace)
print(f"Overall score: {result.overall_score}")
for pattern in result.patterns:
print(f"Found: {pattern.type.value} ({pattern.severity.value})")
```
"""
def __init__(
self,
api_key: str | None = None,
base_url: str = "https://api.minimax.io/anthropic",
model: str = "MiniMax-M2.1",
):
"""
Initialize TraceAnalyzer with M2.1 configuration.
Args:
api_key: MiniMax API key
base_url: API endpoint
model: Model for analysis (M2.1 recommended for best results)
"""
self.model = model
self.client = anthropic.Anthropic(
api_key=api_key or os.environ.get("ANTHROPIC_API_KEY"),
base_url=base_url,
)
def analyze(
self,
trace: ReasoningTrace,
max_tokens: int = 8192,
) -> AnalysisResult:
"""
Analyze a reasoning trace and return detailed analysis.
Args:
trace: The reasoning trace to analyze
max_tokens: Maximum tokens for analysis response
Returns:
AnalysisResult with patterns, scores, and recommendations
"""
# Format trace for analysis
trace_text = self._format_trace_for_analysis(trace)
tool_calls_text = self._format_tool_calls(trace)
prompt = ANALYSIS_PROMPT_TEMPLATE.format(
task=trace.task,
system_prompt=trace.system_prompt,
trace=trace_text,
tool_calls=tool_calls_text,
success=trace.success,
final_response=trace.final_response or "None",
error=trace.error or "None",
)
# Call M2.1 for analysis
response = self.client.messages.create(
model=self.model,
max_tokens=max_tokens,
system=ANALYSIS_SYSTEM_PROMPT,
messages=[{"role": "user", "content": prompt}],
)
# Extract thinking and text from response
analyzer_thinking = ""
analysis_text = ""
for block in response.content:
if block.type == "thinking":
analyzer_thinking = block.thinking
elif block.type == "text":
analysis_text = block.text
# Parse the JSON response
result = self._parse_analysis_response(analysis_text, trace.session_id)
result.analyzer_thinking = analyzer_thinking
result.analyzer_model = self.model
return result
def analyze_batch(
self,
traces: list[ReasoningTrace],
) -> list[AnalysisResult]:
"""Analyze multiple traces and return results."""
return [self.analyze(trace) for trace in traces]
def quick_score(
self,
trace: ReasoningTrace,
) -> float:
"""
Get a quick overall score without full pattern analysis.
Useful for optimization loops where you need fast feedback.
Args:
trace: The reasoning trace to score
Returns:
Overall score from 0-100
"""
quick_prompt = f"""Rate this agent's performance from 0-100 based on its reasoning trace.
Task: {trace.task}
Success: {trace.success}
Turns: {trace.total_turns}
Thinking excerpts:
{self._get_thinking_excerpts(trace, max_chars=2000)}
Respond with ONLY a number from 0-100."""
response = self.client.messages.create(
model=self.model,
max_tokens=100,
messages=[{"role": "user", "content": quick_prompt}],
)
# Extract score from response
for block in response.content:
if block.type == "text":
try:
score = float(block.text.strip())
return min(100, max(0, score))
except ValueError:
pass
return 50.0 # Default middle score if parsing fails
def _format_trace_for_analysis(self, trace: ReasoningTrace) -> str:
"""Format thinking blocks for analysis."""
parts = []
for i, thinking in enumerate(trace.thinking_blocks):
parts.append(f"[Turn {thinking.turn_index}] Thinking:")
parts.append(thinking.content)
parts.append("")
return "\n".join(parts)
def _format_tool_calls(self, trace: ReasoningTrace) -> str:
"""Format tool calls for analysis."""
if not trace.tool_calls:
return "No tool calls made."
parts = []
for tc in trace.tool_calls:
status = "Success" if tc.success else f"Failed: {tc.error}"
parts.append(
f"- {tc.name}({json.dumps(tc.input)}) -> {status}\n"
f" Result: {tc.result[:200] if tc.result else 'None'}..."
)
return "\n".join(parts)
def _get_thinking_excerpts(self, trace: ReasoningTrace, max_chars: int = 2000) -> str:
"""Get excerpts from thinking blocks."""
excerpts = []
remaining = max_chars
for thinking in trace.thinking_blocks:
if remaining <= 0:
break
excerpt = thinking.content[:remaining]
excerpts.append(f"[Turn {thinking.turn_index}]: {excerpt}")
remaining -= len(excerpt) + 20
return "\n\n".join(excerpts)
def _parse_analysis_response(
self,
response_text: str,
trace_id: str,
) -> AnalysisResult:
"""Parse the JSON analysis response from M2.1."""
result = AnalysisResult(trace_id=trace_id)
try:
# Extract JSON from response (may have markdown code blocks)
json_text = response_text
if "```json" in response_text:
json_text = response_text.split("```json")[1].split("```")[0]
elif "```" in response_text:
json_text = response_text.split("```")[1].split("```")[0]
data = json.loads(json_text)
# Parse patterns
for p in data.get("patterns", []):
try:
pattern = Pattern(
type=PatternType(p["type"]),
severity=Severity(p["severity"]),
description=p["description"],
evidence=p.get("evidence", []),
turn_indices=p.get("turn_indices", []),
suggestion=p.get("suggestion", ""),
confidence=p.get("confidence", 0.5),
)
result.patterns.append(pattern)
except (KeyError, ValueError):
continue
# Parse scores
scores = data.get("scores", {})
result.reasoning_clarity = scores.get("reasoning_clarity", 0)
result.goal_adherence = scores.get("goal_adherence", 0)
result.tool_usage_quality = scores.get("tool_usage_quality", 0)
result.error_recovery = scores.get("error_recovery", 0)
result.overall_score = scores.get("overall", 0)
# Parse feedback
result.strengths = data.get("strengths", [])
result.weaknesses = data.get("weaknesses", [])
result.recommendations = data.get("recommendations", [])
except (json.JSONDecodeError, KeyError) as e:
# If parsing fails, try fallback extraction and set reasonable defaults
result = self._fallback_parse_analysis(response_text, trace_id, str(e))
# Warn if score is suspiciously low (likely parsing failure)
if result.overall_score == 0 and not result.patterns:
result.weaknesses.append("WARNING: Analysis may have failed - score is 0 with no patterns detected")
# Try to extract a score from the response text as fallback
fallback_score = self._extract_fallback_score(response_text)
if fallback_score > 0:
result.overall_score = fallback_score
result.recommendations.append(f"Score extracted via fallback: {fallback_score}")
return result
def _fallback_parse_analysis(
self,
response_text: str,
trace_id: str,
error_msg: str,
) -> AnalysisResult:
"""Fallback parsing when JSON extraction fails."""
import re
result = AnalysisResult(trace_id=trace_id)
# Try to extract score from text patterns like "Overall Score: 75" or "overall": 75
score_patterns = [
r'overall["\s:]+(\d+)',
r'Overall Score[:\s]+(\d+)',
r'"overall"[:\s]+(\d+)',
r'Score[:\s]+(\d+)/100',
]
for pattern in score_patterns:
match = re.search(pattern, response_text, re.IGNORECASE)
if match:
result.overall_score = min(100, max(0, int(match.group(1))))
break
# If still no score, use a neutral default (not 0)
if result.overall_score == 0:
result.overall_score = 50 # Neutral default instead of 0
result.recommendations = [
f"Analysis parsing failed ({error_msg}). Using fallback extraction.",
"Consider re-running analysis if results seem inconsistent."
]
result.weaknesses = ["JSON parsing failed - analysis may be incomplete"]
return result
def _extract_fallback_score(self, response_text: str) -> float:
"""Extract a score from response text when JSON parsing fails."""
import re
patterns = [
r'overall["\s:]+(\d+)',
r'Overall Score[:\s]+(\d+)',
r'"overall"[:\s]+(\d+)',
r'(\d+)/100',
r'score[:\s]+(\d+)',
]
for pattern in patterns:
match = re.search(pattern, response_text, re.IGNORECASE)
if match:
score = int(match.group(1))
if 0 <= score <= 100:
return float(score)
return 0.0
def format_analysis_report(analysis: AnalysisResult) -> str:
"""Format an analysis result as a human-readable report."""
lines = [
"=" * 60,
"REASONING TRACE ANALYSIS REPORT",
"=" * 60,
"",
f"Overall Score: {analysis.overall_score}/100",
"",
"Scores:",
f" - Reasoning Clarity: {analysis.reasoning_clarity}/100",
f" - Goal Adherence: {analysis.goal_adherence}/100",
f" - Tool Usage Quality: {analysis.tool_usage_quality}/100",
f" - Error Recovery: {analysis.error_recovery}/100",
"",
]
if analysis.patterns:
lines.append("Detected Patterns:")
for p in analysis.patterns:
lines.append(f"\n [{p.severity.value.upper()}] {p.type.value}")
lines.append(f" {p.description}")
lines.append(f" Suggestion: {p.suggestion}")
if analysis.strengths:
lines.append("\nStrengths:")
for s in analysis.strengths:
lines.append(f" + {s}")
if analysis.weaknesses:
lines.append("\nWeaknesses:")
for w in analysis.weaknesses:
lines.append(f" - {w}")
if analysis.recommendations:
lines.append("\nRecommendations:")
for i, r in enumerate(analysis.recommendations, 1):
lines.append(f" {i}. {r}")
return "\n".join(lines)
reasoning_trace_optimizer/capture.pyShowHide
"""
TraceCapture: Wraps M2.1 API to capture interleaved thinking traces.
This module provides the core functionality for executing agent tasks
through MiniMax M2.1 while capturing all reasoning traces for analysis.
"""
import json
import os
import uuid
from datetime import datetime
from typing import Any, Callable
import anthropic
from reasoning_trace_optimizer.models import (
ReasoningTrace,
ThinkingBlock,
ToolCall,
)
class TraceCapture:
"""
Captures reasoning traces from MiniMax M2.1's interleaved thinking.
This class wraps the Anthropic SDK configured for M2.1 and captures
all thinking blocks, tool calls, and responses during agent execution.
Example:
```python
capture = TraceCapture()
trace = capture.run(
task="What's the weather in San Francisco?",
tools=[weather_tool],
tool_executor=execute_tool
)
print(f"Captured {len(trace.thinking_blocks)} thinking blocks")
```
"""
def __init__(
self,
api_key: str | None = None,
base_url: str = "https://api.minimax.io/anthropic",
model: str = "MiniMax-M2.1",
):
"""
Initialize TraceCapture with M2.1 configuration.
Args:
api_key: MiniMax API key (defaults to ANTHROPIC_API_KEY env var)
base_url: API base URL (international or China endpoint)
model: Model to use (MiniMax-M2.1, MiniMax-M2.1-lightning, MiniMax-M2)
"""
self.model = model
self.client = anthropic.Anthropic(
api_key=api_key or os.environ.get("ANTHROPIC_API_KEY"),
base_url=base_url,
)
def run(
self,
task: str,
system_prompt: str = "You are a helpful assistant.",
tools: list[dict[str, Any]] | None = None,
tool_executor: Callable[[str, dict], str] | None = None,
max_turns: int = 10,
max_tokens: int = 4096,
) -> ReasoningTrace:
"""
Execute a task and capture the full reasoning trace.
Args:
task: The user task/query to execute
system_prompt: System prompt for the agent
tools: List of tool definitions in Anthropic format
tool_executor: Function to execute tool calls (name, input) -> result
max_turns: Maximum conversation turns before stopping
max_tokens: Maximum tokens per response
Returns:
ReasoningTrace containing all thinking blocks, tool calls, and responses
"""
trace = ReasoningTrace(
session_id=str(uuid.uuid4()),
task=task,
system_prompt=system_prompt,
model=self.model,
started_at=datetime.now(),
)
messages = [{"role": "user", "content": task}]
turn = 0
try:
while turn < max_turns:
# Build request parameters
params = {
"model": self.model,
"max_tokens": max_tokens,
"system": system_prompt,
"messages": messages,
}
if tools:
params["tools"] = tools
# Make API call
response = self.client.messages.create(**params)
# Process response content blocks
thinking_blocks, text_blocks, tool_use_blocks = self._process_response(
response, turn, trace
)
# If no tool calls, we're done
if not tool_use_blocks:
trace.final_response = (
text_blocks[0].text if text_blocks else None
)
trace.success = True
break
# Append assistant response to history (CRITICAL for M2.1)
messages.append({"role": "assistant", "content": response.content})
# Execute tools and collect results
tool_results = []
for tool_block in tool_use_blocks:
result = self._execute_tool(
tool_block, tool_executor, turn, trace
)
tool_results.append(
{
"type": "tool_result",
"tool_use_id": tool_block.id,
"content": result,
}
)
# Add tool results to messages
messages.append({"role": "user", "content": tool_results})
turn += 1
trace.total_turns = turn
# Check if we hit max turns without completion
if turn >= max_turns and not trace.success:
trace.success = False
trace.error = f"Reached maximum turns ({max_turns}) without completion"
except Exception as e:
trace.success = False
trace.error = str(e)
trace.completed_at = datetime.now()
return trace
def _process_response(
self,
response: anthropic.types.Message,
turn: int,
trace: ReasoningTrace,
) -> tuple[list, list, list]:
"""Process response content blocks and update trace."""
thinking_blocks = []
text_blocks = []
tool_use_blocks = []
for block in response.content:
if block.type == "thinking":
thinking = ThinkingBlock(
content=block.thinking,
turn_index=turn,
signature=getattr(block, "signature", None),
)
trace.thinking_blocks.append(thinking)
thinking_blocks.append(block)
elif block.type == "text":
text_blocks.append(block)
elif block.type == "tool_use":
tool_use_blocks.append(block)
# Update token count
trace.total_tokens += response.usage.input_tokens + response.usage.output_tokens
return thinking_blocks, text_blocks, tool_use_blocks
def _execute_tool(
self,
tool_block: Any,
executor: Callable[[str, dict], str] | None,
turn: int,
trace: ReasoningTrace,
) -> str:
"""Execute a tool call and record it in the trace."""
tool_call = ToolCall(
id=tool_block.id,
name=tool_block.name,
input=tool_block.input,
turn_index=turn,
)
try:
if executor:
result = executor(tool_block.name, tool_block.input)
else:
result = f"[Mock result for {tool_block.name}]"
tool_call.result = result
tool_call.success = True
except Exception as e:
result = f"Error: {str(e)}"
tool_call.result = result
tool_call.success = False
tool_call.error = str(e)
trace.tool_calls.append(tool_call)
# Link thinking to tool call
if trace.thinking_blocks:
last_thinking = trace.thinking_blocks[-1]
if last_thinking.turn_index == turn:
last_thinking.following_action = f"tool_use:{tool_block.name}"
return result
def run_streaming(
self,
task: str,
system_prompt: str = "You are a helpful assistant.",
tools: list[dict[str, Any]] | None = None,
tool_executor: Callable[[str, dict], str] | None = None,
max_turns: int = 10,
max_tokens: int = 4096,
on_thinking: Callable[[str], None] | None = None,
on_text: Callable[[str], None] | None = None,
on_tool_call: Callable[[str, dict], None] | None = None,
on_error: Callable[[str], None] | None = None,
) -> ReasoningTrace:
"""
Execute a task with streaming output and capture reasoning trace.
Similar to run() but streams thinking and text content in real-time
via callback functions.
Note: For multi-turn tool interactions, the non-streaming run() method
is recommended as it provides more reliable trace capture. Use this
method when you need real-time display of thinking/text content.
Args:
task: The user task/query to execute
system_prompt: System prompt for the agent
tools: List of tool definitions
tool_executor: Function to execute tool calls
max_turns: Maximum conversation turns
max_tokens: Maximum tokens per response
on_thinking: Callback for thinking content chunks
on_text: Callback for text content chunks
on_tool_call: Callback when tool is called (name, input)
on_error: Callback when an error occurs (error message)
Returns:
ReasoningTrace containing the full captured trace
"""
trace = ReasoningTrace(
session_id=str(uuid.uuid4()),
task=task,
system_prompt=system_prompt,
model=self.model,
started_at=datetime.now(),
)
messages = [{"role": "user", "content": task}]
turn = 0
try:
while turn < max_turns:
params = {
"model": self.model,
"max_tokens": max_tokens,
"system": system_prompt,
"messages": messages,
"stream": True,
}
if tools:
params["tools"] = tools
# Collect streamed content
thinking_buffer = ""
text_buffer = ""
tool_use_blocks = []
current_content = []
with self.client.messages.stream(**params) as stream:
for event in stream:
if event.type == "content_block_start":
if hasattr(event, "content_block"):
current_content.append(event.content_block)
elif event.type == "content_block_delta":
if hasattr(event, "delta"):
if event.delta.type == "thinking_delta":
chunk = event.delta.thinking
thinking_buffer += chunk
if on_thinking:
on_thinking(chunk)
elif event.delta.type == "text_delta":
chunk = event.delta.text
text_buffer += chunk
if on_text:
on_text(chunk)
# Get final message for tool_use blocks
final_message = stream.get_final_message()
for block in final_message.content:
if block.type == "tool_use":
tool_use_blocks.append(block)
if on_tool_call:
on_tool_call(block.name, block.input)
# Record thinking block
if thinking_buffer:
trace.thinking_blocks.append(
ThinkingBlock(
content=thinking_buffer,
turn_index=turn,
)
)
# Update tokens
trace.total_tokens += (
final_message.usage.input_tokens + final_message.usage.output_tokens
)
# If no tool calls, we're done
if not tool_use_blocks:
trace.final_response = text_buffer or None
trace.success = True
break
# Append to history
messages.append({"role": "assistant", "content": final_message.content})
# Execute tools
tool_results = []
for tool_block in tool_use_blocks:
result = self._execute_tool(tool_block, tool_executor, turn, trace)
tool_results.append(
{
"type": "tool_result",
"tool_use_id": tool_block.id,
"content": result,
}
)
messages.append({"role": "user", "content": tool_results})
turn += 1
trace.total_turns = turn
if turn >= max_turns and not trace.success:
trace.success = False
trace.error = f"Reached maximum turns ({max_turns})"
except Exception as e:
trace.success = False
trace.error = str(e)
if on_error:
on_error(str(e))
trace.completed_at = datetime.now()
return trace
def format_trace_for_display(trace: ReasoningTrace) -> str:
"""Format a reasoning trace for human-readable display."""
lines = [
f"Session: {trace.session_id}",
f"Task: {trace.task}",
f"Model: {trace.model}",
f"Status: {'Success' if trace.success else 'Failed'}",
f"Turns: {trace.total_turns}",
f"Tokens: {trace.total_tokens}",
"",
"=" * 60,
"REASONING TRACE",
"=" * 60,
]
for i, thinking in enumerate(trace.thinking_blocks):
lines.append(f"\n[Turn {thinking.turn_index}] Thinking:")
lines.append("-" * 40)
lines.append(thinking.content[:500] + "..." if len(thinking.content) > 500 else thinking.content)
# Show tool calls at this turn
turn_tools = trace.get_tool_calls_at_turn(thinking.turn_index)
for tool in turn_tools:
lines.append(f"\n Tool: {tool.name}({json.dumps(tool.input)})")
lines.append(f" Result: {tool.result[:100]}..." if tool.result and len(tool.result) > 100 else f" Result: {tool.result}")
if trace.final_response:
lines.append("\n" + "=" * 60)
lines.append("FINAL RESPONSE")
lines.append("=" * 60)
lines.append(trace.final_response)
if trace.error:
lines.append("\n" + "=" * 60)
lines.append("ERROR")
lines.append("=" * 60)
lines.append(trace.error)
return "\n".join(lines)
reasoning_trace_optimizer/cli.pyShowHide
"""
CLI interface for Reasoning Trace Optimizer.
Provides command-line access to the optimization tools.
"""
import argparse
import json
import sys
from pathlib import Path
from rich.console import Console
from reasoning_trace_optimizer.analyzer import TraceAnalyzer, format_analysis_report
from reasoning_trace_optimizer.capture import TraceCapture, format_trace_for_display
from reasoning_trace_optimizer.loop import OptimizationLoop, LoopConfig
from reasoning_trace_optimizer.skill_generator import SkillGenerator
console = Console()
def cmd_capture(args: argparse.Namespace) -> None:
"""Run a task and capture reasoning trace."""
capture = TraceCapture(
api_key=args.api_key,
base_url=args.base_url,
model=args.model,
)
console.print(f"[cyan]Capturing trace for task: {args.task}[/cyan]")
trace = capture.run(
task=args.task,
system_prompt=args.system_prompt or "You are a helpful assistant.",
max_turns=args.max_turns,
)
# Output trace
output = format_trace_for_display(trace)
if args.output:
Path(args.output).write_text(output)
console.print(f"[green]Trace saved to: {args.output}[/green]")
else:
console.print(output)
def cmd_analyze(args: argparse.Namespace) -> None:
"""Analyze a captured reasoning trace."""
# For now, run capture + analyze together
# In future, could load trace from file
capture = TraceCapture(
api_key=args.api_key,
base_url=args.base_url,
model=args.model,
)
analyzer = TraceAnalyzer(
api_key=args.api_key,
base_url=args.base_url,
model=args.model,
)
console.print(f"[cyan]Capturing and analyzing: {args.task}[/cyan]")
trace = capture.run(
task=args.task,
system_prompt=args.system_prompt or "You are a helpful assistant.",
)
analysis = analyzer.analyze(trace)
# Output analysis
output = format_analysis_report(analysis)
if args.output:
Path(args.output).write_text(output)
console.print(f"[green]Analysis saved to: {args.output}[/green]")
else:
console.print(output)
def cmd_optimize(args: argparse.Namespace) -> None:
"""Run full optimization loop."""
config = LoopConfig(
max_iterations=args.max_iterations,
convergence_threshold=args.convergence_threshold,
min_score_threshold=args.min_score,
save_artifacts=True,
artifacts_dir=args.artifacts_dir,
verbose=True,
)
loop = OptimizationLoop(
config=config,
api_key=args.api_key,
base_url=args.base_url,
model=args.model,
)
console.print(f"[cyan]Starting optimization for: {args.task}[/cyan]")
result = loop.run(
task=args.task,
initial_prompt=args.system_prompt or "You are a helpful assistant.",
)
# Output final prompt
if args.output:
Path(args.output).write_text(result.final_prompt)
console.print(f"[green]Optimized prompt saved to: {args.output}[/green]")
# Generate skill if requested
if args.generate_skill:
generator = SkillGenerator(
api_key=args.api_key,
base_url=args.base_url,
model=args.model,
)
skill_path = generator.generate(
result=result,
skill_name=args.skill_name or "optimized-agent",
output_dir=args.skills_dir,
)
console.print(f"[green]Generated skill at: {skill_path}[/green]")
def cmd_generate_skill(args: argparse.Namespace) -> None:
"""Generate a skill from optimization artifacts."""
# Load summary from artifacts
artifacts_dir = Path(args.artifacts_dir)
summary_path = artifacts_dir / "summary.json"
if not summary_path.exists():
console.print("[red]Error: No optimization summary found. Run optimize first.[/red]")
sys.exit(1)
with open(summary_path) as f:
summary = json.load(f)
# Create minimal loop result from summary
from reasoning_trace_optimizer.models import LoopResult, LoopIteration, ReasoningTrace, AnalysisResult
# Load final prompt
final_prompt_path = artifacts_dir / "final_prompt.txt"
final_prompt = final_prompt_path.read_text() if final_prompt_path.exists() else ""
result = LoopResult(
task=summary.get("task", "Unknown task"),
final_prompt=final_prompt,
total_iterations=summary.get("total_iterations", 0),
initial_score=summary.get("initial_score", 0),
final_score=summary.get("final_score", 0),
improvement_percentage=summary.get("improvement_percentage", 0),
converged=summary.get("converged", False),
)
generator = SkillGenerator(
api_key=args.api_key,
base_url=args.base_url,
model=args.model,
)
skill_path = generator.generate(
result=result,
skill_name=args.skill_name,
output_dir=args.output_dir,
)
console.print(f"[green]Generated skill at: {skill_path}[/green]")
def main() -> None:
"""Main CLI entry point."""
parser = argparse.ArgumentParser(
prog="rto",
description="Reasoning Trace Optimizer - Debug and optimize AI agents using M2.1's interleaved thinking",
)
# Global arguments
parser.add_argument(
"--api-key",
help="MiniMax API key (or set ANTHROPIC_API_KEY env var)",
)
parser.add_argument(
"--base-url",
default="https://api.minimax.io/anthropic",
help="API base URL",
)
parser.add_argument(
"--model",
default="MiniMax-M2.1",
choices=["MiniMax-M2.1", "MiniMax-M2.1-lightning", "MiniMax-M2"],
help="Model to use",
)
subparsers = parser.add_subparsers(dest="command", required=True)
# Capture command
capture_parser = subparsers.add_parser(
"capture",
help="Capture reasoning trace for a task",
)
capture_parser.add_argument("task", help="Task to execute")
capture_parser.add_argument("--system-prompt", "-s", help="System prompt")
capture_parser.add_argument("--max-turns", type=int, default=10)
capture_parser.add_argument("--output", "-o", help="Output file path")
capture_parser.set_defaults(func=cmd_capture)
# Analyze command
analyze_parser = subparsers.add_parser(
"analyze",
help="Capture and analyze reasoning trace",
)
analyze_parser.add_argument("task", help="Task to analyze")
analyze_parser.add_argument("--system-prompt", "-s", help="System prompt")
analyze_parser.add_argument("--output", "-o", help="Output file path")
analyze_parser.set_defaults(func=cmd_analyze)
# Optimize command
optimize_parser = subparsers.add_parser(
"optimize",
help="Run full optimization loop",
)
optimize_parser.add_argument("task", help="Task to optimize for")
optimize_parser.add_argument("--system-prompt", "-s", help="Initial system prompt")
optimize_parser.add_argument("--max-iterations", type=int, default=5)
optimize_parser.add_argument("--convergence-threshold", type=float, default=5.0)
optimize_parser.add_argument("--min-score", type=float, default=80.0)
optimize_parser.add_argument(
"--artifacts-dir",
default="./optimization_artifacts",
help="Directory for artifacts",
)
optimize_parser.add_argument("--output", "-o", help="Output file for final prompt")
optimize_parser.add_argument(
"--generate-skill",
action="store_true",
help="Generate Agent Skill from results",
)
optimize_parser.add_argument("--skill-name", help="Name for generated skill")
optimize_parser.add_argument(
"--skills-dir",
default="./generated_skills",
help="Directory for generated skills",
)
optimize_parser.set_defaults(func=cmd_optimize)
# Generate skill command
skill_parser = subparsers.add_parser(
"generate-skill",
help="Generate skill from optimization artifacts",
)
skill_parser.add_argument("skill_name", help="Name for the skill")
skill_parser.add_argument(
"--artifacts-dir",
default="./optimization_artifacts",
help="Directory with optimization artifacts",
)
skill_parser.add_argument(
"--output-dir",
default="./generated_skills",
help="Output directory for skill",
)
skill_parser.set_defaults(func=cmd_generate_skill)
args = parser.parse_args()
args.func(args)
if __name__ == "__main__":
main()
reasoning_trace_optimizer/loop.pyShowHide
"""
OptimizationLoop: Orchestrates the full capture → analyze → improve → re-run cycle.
This is the main entry point for automated prompt optimization,
running iterative improvements until convergence or max iterations.
"""
import json
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Callable
from rich.console import Console
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.table import Table
from reasoning_trace_optimizer.analyzer import TraceAnalyzer, format_analysis_report
from reasoning_trace_optimizer.capture import TraceCapture, format_trace_for_display
from reasoning_trace_optimizer.models import (
AnalysisResult,
LoopIteration,
LoopResult,
OptimizationResult,
ReasoningTrace,
)
from reasoning_trace_optimizer.optimizer import PromptOptimizer, format_optimization_report
console = Console()
@dataclass
class LoopConfig:
"""Configuration for the optimization loop."""
max_iterations: int = 5
convergence_threshold: float = 3.0 # Stop if improvement < this %
min_score_threshold: float = 75.0 # Stop if score >= this (realistic for complex tasks)
regression_threshold: float = 8.0 # Rollback if score drops by this much
# Scoring weights
success_weight: float = 0.4
score_weight: float = 0.4
error_weight: float = 0.2
# Optimization behavior
use_best_prompt: bool = True # Use best performing prompt, not final
max_prompt_growth: float = 5.0 # Max ratio of new prompt length to original
# Output options
save_artifacts: bool = True
artifacts_dir: str = "./optimization_artifacts"
verbose: bool = True
class OptimizationLoop:
"""
Orchestrates the full optimization cycle.
Runs iterative loops of:
1. Execute agent with current prompt
2. Capture reasoning trace
3. Analyze trace for issues
4. Generate optimized prompt
5. Repeat until convergence
Example:
```python
loop = OptimizationLoop()
result = loop.run(
task="Search for Python tutorials and summarize them",
initial_prompt="You are a helpful research assistant.",
tools=[search_tool],
tool_executor=execute_search
)
print(f"Improved from {result.initial_score} to {result.final_score}")
print(f"Final prompt:\\n{result.final_prompt}")
```
"""
def __init__(
self,
config: LoopConfig | None = None,
api_key: str | None = None,
base_url: str = "https://api.minimax.io/anthropic",
model: str = "MiniMax-M2.1",
):
"""
Initialize the optimization loop.
Args:
config: Loop configuration
api_key: MiniMax API key
base_url: API endpoint
model: Model to use for all components
"""
self.config = config or LoopConfig()
# Initialize components with same configuration
self.capture = TraceCapture(api_key=api_key, base_url=base_url, model=model)
self.analyzer = TraceAnalyzer(api_key=api_key, base_url=base_url, model=model)
self.optimizer = PromptOptimizer(api_key=api_key, base_url=base_url, model=model)
# Create artifacts directory
if self.config.save_artifacts:
Path(self.config.artifacts_dir).mkdir(parents=True, exist_ok=True)
def run(
self,
task: str,
initial_prompt: str,
tools: list[dict[str, Any]] | None = None,
tool_executor: Callable[[str, dict], str] | None = None,
on_iteration: Callable[[LoopIteration], None] | None = None,
) -> LoopResult:
"""
Run the full optimization loop.
Args:
task: The task to optimize for
initial_prompt: Starting system prompt
tools: Tool definitions for the agent
tool_executor: Function to execute tool calls
on_iteration: Optional callback after each iteration
Returns:
LoopResult with all iterations and final optimized prompt
"""
result = LoopResult(task=task, final_prompt=initial_prompt)
current_prompt = initial_prompt
# Track best performing iteration
best_score = 0.0
best_prompt = initial_prompt
best_iteration = 0
consecutive_regressions = 0
if self.config.verbose:
console.print(Panel(
f"[bold]Starting Optimization Loop[/bold]\n\n"
f"Task: {task}\n"
f"Max Iterations: {self.config.max_iterations}\n"
f"Convergence Threshold: {self.config.convergence_threshold}%",
title="Reasoning Trace Optimizer"
))
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
console=console,
disable=not self.config.verbose,
) as progress:
for i in range(self.config.max_iterations):
task_id = progress.add_task(f"Iteration {i + 1}/{self.config.max_iterations}", total=4)
# Step 1: Capture trace
progress.update(task_id, description=f"[cyan]Iteration {i + 1}: Capturing trace...")
trace = self.capture.run(
task=task,
system_prompt=current_prompt,
tools=tools,
tool_executor=tool_executor,
)
progress.advance(task_id)
# Step 2: Analyze trace
progress.update(task_id, description=f"[cyan]Iteration {i + 1}: Analyzing trace...")
analysis = self.analyzer.analyze(trace)
progress.advance(task_id)
# Calculate iteration score
iteration_score = self._calculate_score(trace, analysis)
# Record initial score
if i == 0:
result.initial_score = iteration_score
best_score = iteration_score
best_prompt = current_prompt
# Step 3: Check convergence
should_continue, reason = self._check_convergence(
iteration=i,
score=iteration_score,
prev_score=result.iterations[-1].analysis.overall_score if result.iterations else 0,
best_score=best_score,
consecutive_regressions=consecutive_regressions,
)
# Step 4: Optimize if continuing
optimization = None
if should_continue:
progress.update(task_id, description=f"[cyan]Iteration {i + 1}: Optimizing prompt...")
optimization = self.optimizer.optimize(
original_prompt=current_prompt,
analysis=analysis,
trace=trace,
)
# Check for excessive prompt growth
new_prompt = optimization.optimized_prompt
if len(new_prompt) > len(initial_prompt) * self.config.max_prompt_growth:
if self.config.verbose:
console.print(f"[yellow]Warning: Prompt grew too large ({len(new_prompt)} chars), limiting growth[/yellow]")
# Keep the current prompt instead of the bloated one
new_prompt = current_prompt
current_prompt = new_prompt
progress.advance(task_id)
# Track best performing iteration AFTER optimization
# This ensures we capture the optimized prompt, not the input prompt
if iteration_score > best_score:
best_score = iteration_score
# Use the optimized prompt if available, otherwise the current prompt
if optimization and optimization.optimized_prompt != initial_prompt:
best_prompt = optimization.optimized_prompt
else:
best_prompt = current_prompt
best_iteration = i + 1
consecutive_regressions = 0
elif iteration_score < best_score - self.config.regression_threshold:
consecutive_regressions += 1
if self.config.verbose:
console.print(f"[yellow]Warning: Score regressed from {best_score:.1f} to {iteration_score:.1f}[/yellow]")
# Record iteration
iteration = LoopIteration(
iteration=i + 1,
trace=trace,
analysis=analysis,
optimization=optimization,
task_completed=trace.success or False,
error_count=len([tc for tc in trace.tool_calls if not tc.success]),
token_usage=trace.total_tokens,
)
result.iterations.append(iteration)
# Callback
if on_iteration:
on_iteration(iteration)
# Print iteration summary
if self.config.verbose:
self._print_iteration_summary(iteration)
# Save artifacts
if self.config.save_artifacts:
self._save_iteration_artifacts(iteration, i + 1)
# Check if we should stop
if not should_continue:
if self.config.verbose:
console.print(f"\n[green]Stopping: {reason}[/green]")
result.converged = True
break
progress.remove_task(task_id)
# Finalize result - use best prompt if configured
if self.config.use_best_prompt and best_score > result.iterations[-1].analysis.overall_score:
result.final_prompt = best_prompt
result.final_score = best_score
if self.config.verbose:
console.print(f"[green]Using best prompt from iteration {best_iteration} (score: {best_score:.1f})[/green]")
else:
result.final_prompt = current_prompt
result.final_score = result.iterations[-1].analysis.overall_score if result.iterations else 0
result.total_iterations = len(result.iterations)
result.improvement_percentage = (
(result.final_score - result.initial_score) / max(result.initial_score, 1) * 100
)
# Warn if prompt was never successfully optimized
if result.final_prompt == initial_prompt:
if self.config.verbose:
console.print(
"[yellow]Warning: Final prompt unchanged from initial. "
"Optimization may have failed to parse model responses.[/yellow]"
)
# Check if any iteration actually produced a different prompt
any_optimized = any(
i.optimization and i.optimization.optimized_prompt != initial_prompt
for i in result.iterations
if i.optimization
)
if not any_optimized:
console.print(
"[yellow]No successful prompt optimizations were extracted. "
"Check artifacts for raw optimizer responses.[/yellow]"
)
# Print final summary
if self.config.verbose:
self._print_final_summary(result)
# Save final artifacts
if self.config.save_artifacts:
self._save_final_artifacts(result)
return result
def run_single(
self,
task: str,
prompt: str,
tools: list[dict[str, Any]] | None = None,
tool_executor: Callable[[str, dict], str] | None = None,
) -> tuple[ReasoningTrace, AnalysisResult]:
"""
Run a single capture + analysis cycle (no optimization).
Useful for debugging or when you just want analysis without
automatic optimization.
Returns:
Tuple of (trace, analysis)
"""
trace = self.capture.run(
task=task,
system_prompt=prompt,
tools=tools,
tool_executor=tool_executor,
)
analysis = self.analyzer.analyze(trace)
return trace, analysis
def _calculate_score(
self,
trace: ReasoningTrace,
analysis: AnalysisResult,
) -> float:
"""Calculate weighted score from trace and analysis."""
success_score = 100 if trace.success else 0
error_penalty = len([tc for tc in trace.tool_calls if not tc.success]) * 10
weighted = (
success_score * self.config.success_weight
+ analysis.overall_score * self.config.score_weight
- error_penalty * self.config.error_weight
)
return max(0, min(100, weighted))
def _check_convergence(
self,
iteration: int,
score: float,
prev_score: float,
best_score: float = 0.0,
consecutive_regressions: int = 0,
) -> tuple[bool, str]:
"""Check if optimization should continue."""
# Check score threshold
if score >= self.config.min_score_threshold:
return False, f"Score {score:.1f} >= threshold {self.config.min_score_threshold}"
# Check for consecutive regressions (stop if we've regressed twice in a row)
if consecutive_regressions >= 2:
return False, f"Consecutive regressions detected (best was {best_score:.1f})"
# Check improvement threshold (after first iteration)
if iteration > 0:
improvement = score - prev_score
if abs(improvement) < self.config.convergence_threshold and score >= prev_score:
return False, f"Converged (improvement {improvement:.1f}% < threshold)"
# Check max iterations
if iteration >= self.config.max_iterations - 1:
return False, f"Reached max iterations ({self.config.max_iterations})"
return True, ""
def _print_iteration_summary(self, iteration: LoopIteration) -> None:
"""Print summary of an iteration."""
table = Table(title=f"Iteration {iteration.iteration} Summary")
table.add_column("Metric", style="cyan")
table.add_column("Value", style="green")
table.add_row("Task Completed", "Yes" if iteration.task_completed else "No")
table.add_row("Overall Score", f"{iteration.analysis.overall_score:.1f}/100")
table.add_row("Patterns Found", str(len(iteration.analysis.patterns)))
table.add_row("Tool Errors", str(iteration.error_count))
table.add_row("Token Usage", str(iteration.token_usage))
if iteration.optimization:
table.add_row(
"Predicted Improvement",
f"{iteration.optimization.predicted_improvement}%"
)
console.print(table)
def _print_final_summary(self, result: LoopResult) -> None:
"""Print final optimization summary."""
console.print("\n")
panel_content = (
f"[bold]Iterations:[/bold] {result.total_iterations}\n"
f"[bold]Converged:[/bold] {'Yes' if result.converged else 'No'}\n"
f"[bold]Initial Score:[/bold] {result.initial_score:.1f}\n"
f"[bold]Final Score:[/bold] {result.final_score:.1f}\n"
f"[bold]Improvement:[/bold] {result.improvement_percentage:+.1f}%"
)
console.print(Panel(panel_content, title="[green]Optimization Complete[/green]"))
def _save_iteration_artifacts(self, iteration: LoopIteration, num: int) -> None:
"""Save iteration artifacts to disk."""
base_path = Path(self.config.artifacts_dir) / f"iteration_{num}"
base_path.mkdir(exist_ok=True)
# Save trace
with open(base_path / "trace.txt", "w") as f:
f.write(format_trace_for_display(iteration.trace))
# Save analysis
with open(base_path / "analysis.txt", "w") as f:
f.write(format_analysis_report(iteration.analysis))
# Save optimization if present
if iteration.optimization:
with open(base_path / "optimization.txt", "w") as f:
f.write(format_optimization_report(iteration.optimization))
with open(base_path / "optimized_prompt.txt", "w") as f:
f.write(iteration.optimization.optimized_prompt)
def _save_final_artifacts(self, result: LoopResult) -> None:
"""Save final optimization artifacts."""
base_path = Path(self.config.artifacts_dir)
# Save final prompt
with open(base_path / "final_prompt.txt", "w") as f:
f.write(result.final_prompt)
# Save summary JSON
summary = {
"task": result.task,
"total_iterations": result.total_iterations,
"converged": result.converged,
"initial_score": result.initial_score,
"final_score": result.final_score,
"improvement_percentage": result.improvement_percentage,
"timestamp": datetime.now().isoformat(),
}
with open(base_path / "summary.json", "w") as f:
json.dump(summary, f, indent=2)
def run_quick_optimization(
task: str,
initial_prompt: str,
tools: list[dict[str, Any]] | None = None,
tool_executor: Callable[[str, dict], str] | None = None,
max_iterations: int = 3,
) -> str:
"""
Quick helper function for one-shot optimization.
Returns the optimized prompt directly.
"""
config = LoopConfig(max_iterations=max_iterations, verbose=False)
loop = OptimizationLoop(config=config)
result = loop.run(task, initial_prompt, tools, tool_executor)
return result.final_prompt
reasoning_trace_optimizer/models.pyShowHide
"""
Core data models for reasoning trace optimization.
"""
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any
class PatternType(Enum):
"""Types of patterns detected in reasoning traces."""
CONTEXT_DEGRADATION = "context_degradation"
TOOL_CONFUSION = "tool_confusion"
INSTRUCTION_DRIFT = "instruction_drift"
HALLUCINATION = "hallucination"
INCOMPLETE_REASONING = "incomplete_reasoning"
TOOL_MISUSE = "tool_misuse"
GOAL_ABANDONMENT = "goal_abandonment"
CIRCULAR_REASONING = "circular_reasoning"
PREMATURE_CONCLUSION = "premature_conclusion"
MISSING_VALIDATION = "missing_validation"
class Severity(Enum):
"""Severity levels for detected patterns."""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class ThinkingBlock:
"""A single thinking/reasoning block from the model."""
content: str
turn_index: int
timestamp: datetime = field(default_factory=datetime.now)
token_count: int = 0
signature: str | None = None # M2.1 thinking signature
# Context at time of thinking
preceding_tool_call: str | None = None
preceding_tool_result: str | None = None
following_action: str | None = None # tool_use, text, or end_turn
@dataclass
class ToolCall:
"""A tool call made by the agent."""
id: str
name: str
input: dict[str, Any]
turn_index: int
result: str | None = None
success: bool | None = None
error: str | None = None
@dataclass
class ReasoningTrace:
"""Complete reasoning trace for an agent session."""
session_id: str
task: str
system_prompt: str
thinking_blocks: list[ThinkingBlock] = field(default_factory=list)
tool_calls: list[ToolCall] = field(default_factory=list)
final_response: str | None = None
# Metadata
model: str = "MiniMax-M2.1"
total_turns: int = 0
total_tokens: int = 0
success: bool | None = None
error: str | None = None
started_at: datetime = field(default_factory=datetime.now)
completed_at: datetime | None = None
def get_thinking_at_turn(self, turn: int) -> ThinkingBlock | None:
"""Get thinking block at specific turn."""
for block in self.thinking_blocks:
if block.turn_index == turn:
return block
return None
def get_tool_calls_at_turn(self, turn: int) -> list[ToolCall]:
"""Get all tool calls at specific turn."""
return [tc for tc in self.tool_calls if tc.turn_index == turn]
@dataclass
class Pattern:
"""A detected pattern in reasoning traces."""
type: PatternType
severity: Severity
description: str
evidence: list[str] # Excerpts from thinking blocks
turn_indices: list[int]
suggestion: str
confidence: float # 0.0 to 1.0
@dataclass
class AnalysisResult:
"""Result of analyzing a reasoning trace."""
trace_id: str
patterns: list[Pattern] = field(default_factory=list)
# Scores (0-100)
reasoning_clarity: float = 0.0
goal_adherence: float = 0.0
tool_usage_quality: float = 0.0
error_recovery: float = 0.0
overall_score: float = 0.0
# Feedback
strengths: list[str] = field(default_factory=list)
weaknesses: list[str] = field(default_factory=list)
recommendations: list[str] = field(default_factory=list)
# Analysis metadata
analyzer_model: str = "MiniMax-M2.1"
analyzer_thinking: str = "" # The analyzer's own reasoning
@dataclass
class PromptDiff:
"""Difference between original and optimized prompt."""
section: str # e.g., "system_prompt", "tool_description", "instruction"
original: str
optimized: str
reason: str
@dataclass
class OptimizationResult:
"""Result of prompt optimization."""
original_prompt: str
optimized_prompt: str
diffs: list[PromptDiff] = field(default_factory=list)
# Improvement predictions
predicted_improvement: float = 0.0 # Percentage
confidence: float = 0.0
# Optimizer reasoning
optimizer_thinking: str = ""
key_changes: list[str] = field(default_factory=list)
@dataclass
class LoopIteration:
"""Single iteration of the optimization loop."""
iteration: int
trace: ReasoningTrace
analysis: AnalysisResult
optimization: OptimizationResult | None
# Metrics
task_completed: bool = False
error_count: int = 0
token_usage: int = 0
@dataclass
class LoopResult:
"""Result of running the full optimization loop."""
task: str
iterations: list[LoopIteration] = field(default_factory=list)
# Final state
final_prompt: str = ""
converged: bool = False
total_iterations: int = 0
# Improvement metrics
initial_score: float = 0.0
final_score: float = 0.0
improvement_percentage: float = 0.0
# Generated artifacts
generated_skill_path: str | None = None
reasoning_trace_optimizer/optimizer.pyShowHide
"""
PromptOptimizer: Generates improved prompts based on trace analysis.
Uses M2.1 to synthesize analysis results into concrete prompt improvements,
with full reasoning transparency via interleaved thinking.
"""
import json
import os
from typing import Any
import anthropic
from reasoning_trace_optimizer.models import (
AnalysisResult,
OptimizationResult,
PromptDiff,
ReasoningTrace,
)
OPTIMIZER_SYSTEM_PROMPT = """You are an expert prompt engineer specializing in AI agent optimization.
Your task is to improve agent prompts based on reasoning trace analysis.
You have access to:
1. The original prompt that was used
2. Analysis of how the agent reasoned (its thinking trace)
3. Detected patterns and issues
4. Specific recommendations
Your goal is to create an IMPROVED prompt that:
- Addresses identified weaknesses
- Maintains existing strengths
- Prevents detected failure patterns
- Improves clarity and specificity
When optimizing, consider:
- Adding explicit guardrails for common failure modes
- Clarifying ambiguous instructions
- Adding examples for complex behaviors
- Restructuring for better context positioning
- Adding validation steps where missing
Provide the optimized prompt with clear explanations of changes."""
OPTIMIZATION_PROMPT_TEMPLATE = """Optimize the following agent prompt based on trace analysis:
## Original Task
{task}
## Original System Prompt
```
{original_prompt}
```
## Analysis Results
### Overall Score: {overall_score}/100
### Detected Patterns
{patterns}
### Weaknesses
{weaknesses}
### Recommendations
{recommendations}
### Analyzer's Reasoning
{analyzer_thinking}
---
Provide your optimization as JSON:
```json
{{
"optimized_prompt": "<the full improved prompt>",
"diffs": [
{{
"section": "<which part changed, e.g., 'instructions', 'guardrails', 'examples'>",
"original": "<original text or 'N/A' if new>",
"optimized": "<new/changed text>",
"reason": "<why this change helps>"
}}
],
"key_changes": [
"<summary of major change 1>",
"<summary of major change 2>"
],
"predicted_improvement": 15,
"confidence": 0.75
}}
```
Think carefully about what changes will have the biggest impact on agent performance."""
class PromptOptimizer:
"""
Optimizes agent prompts based on reasoning trace analysis.
Uses M2.1's interleaved thinking to generate thoughtful improvements
with full transparency into the optimization reasoning.
Example:
```python
optimizer = PromptOptimizer()
result = optimizer.optimize(
original_prompt=system_prompt,
analysis=analysis_result,
trace=reasoning_trace
)
print(f"Predicted improvement: {result.predicted_improvement}%")
print(f"New prompt:\\n{result.optimized_prompt}")
```
"""
def __init__(
self,
api_key: str | None = None,
base_url: str = "https://api.minimax.io/anthropic",
model: str = "MiniMax-M2.1",
):
"""
Initialize PromptOptimizer with M2.1 configuration.
Args:
api_key: MiniMax API key
base_url: API endpoint
model: Model for optimization
"""
self.model = model
self.client = anthropic.Anthropic(
api_key=api_key or os.environ.get("ANTHROPIC_API_KEY"),
base_url=base_url,
)
def optimize(
self,
original_prompt: str,
analysis: AnalysisResult,
trace: ReasoningTrace | None = None,
max_tokens: int = 8192,
) -> OptimizationResult:
"""
Generate an optimized prompt based on analysis.
Args:
original_prompt: The original system prompt to improve
analysis: Analysis results from TraceAnalyzer
trace: Optional original trace for additional context
max_tokens: Maximum tokens for response
Returns:
OptimizationResult with new prompt and change details
"""
# Format analysis for prompt
patterns_text = self._format_patterns(analysis)
weaknesses_text = "\n".join(f"- {w}" for w in analysis.weaknesses)
recommendations_text = "\n".join(f"- {r}" for r in analysis.recommendations)
prompt = OPTIMIZATION_PROMPT_TEMPLATE.format(
task=trace.task if trace else "Unknown task",
original_prompt=original_prompt,
overall_score=analysis.overall_score,
patterns=patterns_text,
weaknesses=weaknesses_text or "None identified",
recommendations=recommendations_text or "None provided",
analyzer_thinking=analysis.analyzer_thinking[:2000] if analysis.analyzer_thinking else "Not available",
)
# Call M2.1 for optimization
response = self.client.messages.create(
model=self.model,
max_tokens=max_tokens,
system=OPTIMIZER_SYSTEM_PROMPT,
messages=[{"role": "user", "content": prompt}],
)
# Extract thinking and response
optimizer_thinking = ""
optimization_text = ""
for block in response.content:
if block.type == "thinking":
optimizer_thinking = block.thinking
elif block.type == "text":
optimization_text = block.text
# Parse the response
result = self._parse_optimization_response(optimization_text, original_prompt)
result.optimizer_thinking = optimizer_thinking
return result
def optimize_iterative(
self,
original_prompt: str,
analyses: list[AnalysisResult],
traces: list[ReasoningTrace],
) -> OptimizationResult:
"""
Optimize based on multiple analysis iterations.
Synthesizes patterns across multiple runs for more robust improvements.
Args:
original_prompt: The original system prompt
analyses: List of analysis results from multiple runs
traces: Corresponding reasoning traces
Returns:
OptimizationResult incorporating learnings from all iterations
"""
# Aggregate patterns across all analyses
all_patterns = []
all_weaknesses = []
all_recommendations = []
avg_score = 0
for analysis in analyses:
all_patterns.extend(analysis.patterns)
all_weaknesses.extend(analysis.weaknesses)
all_recommendations.extend(analysis.recommendations)
avg_score += analysis.overall_score
avg_score /= len(analyses) if analyses else 1
# Create aggregated analysis
aggregated = AnalysisResult(
trace_id="aggregated",
patterns=all_patterns,
overall_score=avg_score,
weaknesses=list(set(all_weaknesses)), # Deduplicate
recommendations=list(set(all_recommendations)),
)
# Optimize based on aggregated analysis
return self.optimize(
original_prompt=original_prompt,
analysis=aggregated,
trace=traces[0] if traces else None,
)
def suggest_tool_improvements(
self,
tools: list[dict[str, Any]],
analysis: AnalysisResult,
trace: ReasoningTrace,
) -> dict[str, str]:
"""
Suggest improvements for tool definitions based on analysis.
Args:
tools: Original tool definitions
analysis: Analysis results
trace: Original reasoning trace
Returns:
Dict mapping tool names to suggested description improvements
"""
tool_issues = [
p for p in analysis.patterns
if p.type.value in ("tool_confusion", "tool_misuse")
]
if not tool_issues:
return {}
prompt = f"""Based on these tool usage issues:
{self._format_patterns_for_tools(tool_issues)}
And the original tool definitions:
{json.dumps(tools, indent=2)}
Suggest improved tool descriptions. Respond as JSON:
```json
{{
"tool_name": "improved description that addresses the confusion"
}}
```"""
response = self.client.messages.create(
model=self.model,
max_tokens=2048,
messages=[{"role": "user", "content": prompt}],
)
for block in response.content:
if block.type == "text":
try:
text = block.text
if "```json" in text:
text = text.split("```json")[1].split("```")[0]
return json.loads(text)
except json.JSONDecodeError:
pass
return {}
def _format_patterns(self, analysis: AnalysisResult) -> str:
"""Format patterns for optimization prompt."""
if not analysis.patterns:
return "No significant patterns detected."
parts = []
for p in analysis.patterns:
parts.append(
f"[{p.severity.value.upper()}] {p.type.value}\n"
f" Description: {p.description}\n"
f" Evidence: {', '.join(p.evidence[:2])}\n"
f" Suggestion: {p.suggestion}"
)
return "\n\n".join(parts)
def _format_patterns_for_tools(self, patterns: list) -> str:
"""Format tool-related patterns."""
return "\n".join(
f"- {p.type.value}: {p.description}" for p in patterns
)
def _parse_optimization_response(
self,
response_text: str,
original_prompt: str,
) -> OptimizationResult:
"""Parse the JSON optimization response with fallback extraction."""
result = OptimizationResult(
original_prompt=original_prompt,
optimized_prompt=original_prompt, # Default to original if parsing fails
)
try:
json_text = response_text
if "```json" in response_text:
json_text = response_text.split("```json")[1].split("```")[0]
elif "```" in response_text:
json_text = response_text.split("```")[1].split("```")[0]
data = json.loads(json_text)
result.optimized_prompt = data.get("optimized_prompt", original_prompt)
result.predicted_improvement = data.get("predicted_improvement", 0)
result.confidence = data.get("confidence", 0.5)
result.key_changes = data.get("key_changes", [])
# Parse diffs
for d in data.get("diffs", []):
diff = PromptDiff(
section=d.get("section", "unknown"),
original=d.get("original", ""),
optimized=d.get("optimized", ""),
reason=d.get("reason", ""),
)
result.diffs.append(diff)
except (json.JSONDecodeError, KeyError) as e:
# Fallback: try to extract optimized_prompt directly from response
extracted_prompt = self._fallback_extract_prompt(response_text)
if extracted_prompt and extracted_prompt != original_prompt:
result.optimized_prompt = extracted_prompt
result.key_changes = [f"JSON parsing failed ({type(e).__name__}), extracted prompt via fallback"]
result.confidence = 0.3 # Lower confidence for fallback extraction
else:
result.key_changes = [f"Optimization parsing failed ({type(e).__name__}) - using original prompt"]
return result
def _fallback_extract_prompt(self, response_text: str) -> str | None:
"""
Fallback method to extract optimized prompt when JSON parsing fails.
Tries multiple strategies to find the prompt content.
"""
import re
# Strategy 1: Look for "optimized_prompt": "..." pattern
match = re.search(r'"optimized_prompt"\s*:\s*"([^"]+)"', response_text, re.DOTALL)
if match:
# Unescape the string
return match.group(1).replace('\\n', '\n').replace('\\"', '"')
# Strategy 2: Look for content between specific markers
markers = [
('## Optimized Prompt', '##'),
('**Optimized Prompt**', '**'),
('OPTIMIZED PROMPT:', '\n\n'),
('Here is the improved prompt:', '\n\n---'),
]
for start_marker, end_marker in markers:
if start_marker in response_text:
start_idx = response_text.find(start_marker) + len(start_marker)
remaining = response_text[start_idx:].strip()
if end_marker in remaining:
end_idx = remaining.find(end_marker)
extracted = remaining[:end_idx].strip()
if len(extracted) > 50: # Minimum length check
return extracted
# Strategy 3: Look for a substantial code block that might be the prompt
code_blocks = re.findall(r'```(?:text|markdown)?\n(.*?)```', response_text, re.DOTALL)
for block in code_blocks:
# Skip JSON blocks, look for prose blocks that could be prompts
if not block.strip().startswith('{') and len(block) > 100:
return block.strip()
return None
def format_optimization_report(result: OptimizationResult) -> str:
"""Format an optimization result as a human-readable report."""
lines = [
"=" * 60,
"PROMPT OPTIMIZATION REPORT",
"=" * 60,
"",
f"Predicted Improvement: {result.predicted_improvement}%",
f"Confidence: {result.confidence * 100:.0f}%",
"",
]
if result.key_changes:
lines.append("Key Changes:")
for change in result.key_changes:
lines.append(f" - {change}")
lines.append("")
if result.diffs:
lines.append("Detailed Changes:")
for diff in result.diffs:
lines.append(f"\n [{diff.section}]")
if diff.original and diff.original != "N/A":
lines.append(f" Before: {diff.original[:100]}...")
lines.append(f" After: {diff.optimized[:100]}...")
lines.append(f" Reason: {diff.reason}")
lines.extend([
"",
"=" * 60,
"OPTIMIZED PROMPT",
"=" * 60,
result.optimized_prompt,
])
return "\n".join(lines)
reasoning_trace_optimizer/skill_generator.pyShowHide
"""
SkillGenerator: Converts optimization insights into shareable Agent Skills.
Transforms the learnings from optimization loops into reusable skills
following the Agent Skills template format.
"""
import json
import os
from datetime import datetime
from pathlib import Path
from typing import Any
import anthropic
from reasoning_trace_optimizer.models import (
AnalysisResult,
LoopResult,
Pattern,
PatternType,
)
SKILL_TEMPLATE = '''---
name: {skill_name}
description: "{description}"
---
# {title}
{intro}
## When to Activate
{activation}
## Core Concepts
{concepts}
## Patterns to Avoid
{anti_patterns}
## Recommended Practices
{practices}
## Guidelines
{guidelines}
## Examples
{examples}
---
## Skill Metadata
**Generated**: {date}
**Source**: Reasoning Trace Optimizer
**Optimization Iterations**: {iterations}
**Score Improvement**: {initial_score:.1f} → {final_score:.1f} (+{improvement:.1f}%)
'''
GENERATOR_SYSTEM_PROMPT = """You are an expert at converting agent optimization insights into reusable skills.
Your task is to analyze optimization results and generate a shareable Agent Skill that
captures the learnings so other developers can benefit.
The skill should:
1. Describe WHEN to use these learnings (activation triggers)
2. Explain the PATTERNS to avoid (anti-patterns found)
3. Provide CONCRETE practices that improved performance
4. Give VERIFIABLE guidelines (things that can be checked)
5. Include EXAMPLES showing before/after improvements
Write in a clear, direct style. Focus on actionable guidance, not theory."""
def _format_list_to_markdown(items: list | str) -> str:
"""Convert a list to markdown bullet points."""
if isinstance(items, str):
return items
if not items:
return ""
import re
formatted = []
for item in items:
# Strip any existing leading bullet points/dashes to avoid duplication
cleaned = re.sub(r'^[-*•]\s*', '', str(item).strip())
formatted.append(f"- {cleaned}")
return "\n".join(formatted)
def _format_numbered_list_to_markdown(items: list | str) -> str:
"""Convert a list to markdown numbered list."""
if isinstance(items, str):
return items
if not items:
return ""
import re
formatted = []
for i, item in enumerate(items):
# Strip any existing leading numbers (e.g., "1. ", "2. ") to avoid duplication
cleaned = re.sub(r'^\d+\.\s*', '', str(item).strip())
formatted.append(f"{i+1}. {cleaned}")
return "\n".join(formatted)
def _format_examples_to_markdown(examples: list | str) -> str:
"""Convert example dicts to markdown format."""
if isinstance(examples, str):
return examples
if not examples:
return ""
parts = []
for i, ex in enumerate(examples):
if isinstance(ex, dict):
parts.append(f"### Example {i+1}: {ex.get('context', 'Scenario')}")
if ex.get('before'):
parts.append(f"\n**Before:**\n```\n{ex['before']}\n```")
if ex.get('after'):
parts.append(f"\n**After:**\n```\n{ex['after']}\n```")
if ex.get('improvement'):
parts.append(f"\n**Improvement:** {ex['improvement']}")
parts.append("")
else:
parts.append(f"- {ex}")
return "\n".join(parts)
class SkillGenerator:
"""
Generates shareable Agent Skills from optimization results.
Converts the learnings from optimization loops into the standard
Agent Skills format for sharing with other developers.
Example:
```python
generator = SkillGenerator()
skill_path = generator.generate(
result=loop_result,
skill_name="web-search-agent",
output_dir="./generated_skills"
)
print(f"Generated skill at: {skill_path}")
```
"""
def __init__(
self,
api_key: str | None = None,
base_url: str = "https://api.minimax.io/anthropic",
model: str = "MiniMax-M2.1",
):
"""
Initialize SkillGenerator.
Args:
api_key: MiniMax API key
base_url: API endpoint
model: Model for skill generation
"""
self.model = model
self.client = anthropic.Anthropic(
api_key=api_key or os.environ.get("ANTHROPIC_API_KEY"),
base_url=base_url,
)
def generate(
self,
result: LoopResult,
skill_name: str,
output_dir: str = "./generated_skills",
title: str | None = None,
) -> str:
"""
Generate an Agent Skill from optimization results.
Args:
result: The optimization loop result
skill_name: Name for the skill (lowercase-with-hyphens)
output_dir: Directory to save the skill
title: Optional human-readable title
Returns:
Path to the generated SKILL.md file
"""
# Extract insights from all iterations
all_patterns = self._collect_patterns(result)
all_recommendations = self._collect_recommendations(result)
key_changes = self._collect_key_changes(result)
# Generate skill content using M2.1
content = self._generate_skill_content(
task=result.task,
patterns=all_patterns,
recommendations=all_recommendations,
key_changes=key_changes,
initial_prompt=result.iterations[0].trace.system_prompt if result.iterations else "",
final_prompt=result.final_prompt,
)
# Format content - convert lists to markdown
formatted_content = {
"activation": _format_list_to_markdown(content.get("activation", "")),
"concepts": _format_list_to_markdown(content.get("concepts", "")),
"anti_patterns": _format_list_to_markdown(content.get("anti_patterns", "")),
"practices": _format_list_to_markdown(content.get("practices", "")),
"guidelines": _format_numbered_list_to_markdown(content.get("guidelines", "")),
"examples": _format_examples_to_markdown(content.get("examples", "")),
}
# Format using template
skill_content = SKILL_TEMPLATE.format(
skill_name=skill_name,
description=content.get("description", f"Optimized practices for {skill_name}"),
title=title or content.get("title", skill_name.replace("-", " ").title()),
intro=content.get("intro", ""),
activation=formatted_content["activation"],
concepts=formatted_content["concepts"],
anti_patterns=formatted_content["anti_patterns"],
practices=formatted_content["practices"],
guidelines=formatted_content["guidelines"],
examples=formatted_content["examples"],
date=datetime.now().strftime("%Y-%m-%d"),
iterations=result.total_iterations,
initial_score=result.initial_score,
final_score=result.final_score,
improvement=result.improvement_percentage,
)
# Save skill
skill_dir = Path(output_dir) / skill_name
skill_dir.mkdir(parents=True, exist_ok=True)
skill_path = skill_dir / "SKILL.md"
with open(skill_path, "w") as f:
f.write(skill_content)
# Save optimization data as reference
self._save_references(skill_dir, result, content)
return str(skill_path)
def generate_from_analysis(
self,
analyses: list[AnalysisResult],
skill_name: str,
task_description: str,
output_dir: str = "./generated_skills",
) -> str:
"""
Generate a skill from multiple analysis results (without full loop).
Useful when you have analysis data but didn't run the full optimization loop.
Args:
analyses: List of analysis results
skill_name: Name for the skill
task_description: Description of the task context
output_dir: Output directory
Returns:
Path to generated skill
"""
# Aggregate patterns and recommendations
all_patterns = []
all_recommendations = []
for analysis in analyses:
all_patterns.extend(analysis.patterns)
all_recommendations.extend(analysis.recommendations)
content = self._generate_skill_content(
task=task_description,
patterns=all_patterns,
recommendations=list(set(all_recommendations)),
key_changes=[],
initial_prompt="",
final_prompt="",
)
# Calculate average score
avg_score = sum(a.overall_score for a in analyses) / len(analyses) if analyses else 0
skill_content = SKILL_TEMPLATE.format(
skill_name=skill_name,
description=content.get("description", f"Learnings for {skill_name}"),
title=content.get("title", skill_name.replace("-", " ").title()),
intro=content.get("intro", ""),
activation=content.get("activation", ""),
concepts=content.get("concepts", ""),
anti_patterns=content.get("anti_patterns", ""),
practices=content.get("practices", ""),
guidelines=content.get("guidelines", ""),
examples=content.get("examples", ""),
date=datetime.now().strftime("%Y-%m-%d"),
iterations=len(analyses),
initial_score=avg_score,
final_score=avg_score,
improvement=0,
)
skill_dir = Path(output_dir) / skill_name
skill_dir.mkdir(parents=True, exist_ok=True)
skill_path = skill_dir / "SKILL.md"
with open(skill_path, "w") as f:
f.write(skill_content)
return str(skill_path)
def _collect_patterns(self, result: LoopResult) -> list[Pattern]:
"""Collect all unique patterns from iterations."""
patterns = []
seen = set()
for iteration in result.iterations:
for pattern in iteration.analysis.patterns:
key = (pattern.type, pattern.description[:50])
if key not in seen:
patterns.append(pattern)
seen.add(key)
return patterns
def _collect_recommendations(self, result: LoopResult) -> list[str]:
"""Collect all unique recommendations."""
recommendations = []
seen = set()
for iteration in result.iterations:
for rec in iteration.analysis.recommendations:
if rec not in seen:
recommendations.append(rec)
seen.add(rec)
return recommendations
def _collect_key_changes(self, result: LoopResult) -> list[str]:
"""Collect all key changes from optimizations."""
changes = []
for iteration in result.iterations:
if iteration.optimization:
changes.extend(iteration.optimization.key_changes)
return changes
def _generate_skill_content(
self,
task: str,
patterns: list[Pattern],
recommendations: list[str],
key_changes: list[str],
initial_prompt: str,
final_prompt: str,
) -> dict[str, str]:
"""Use M2.1 to generate skill content sections."""
patterns_text = "\n".join(
f"- [{p.severity.value}] {p.type.value}: {p.description}"
for p in patterns
)
recommendations_text = "\n".join(f"- {r}" for r in recommendations)
changes_text = "\n".join(f"- {c}" for c in key_changes)
prompt = f"""Generate an Agent Skill based on these optimization insights:
## Task Context
{task}
## Patterns Detected (Anti-patterns to avoid)
{patterns_text or "No significant patterns detected"}
## Recommendations from Analysis
{recommendations_text or "No specific recommendations"}
## Key Changes That Improved Performance
{changes_text or "No recorded changes"}
## Prompt Evolution
Initial: {initial_prompt[:500] if initial_prompt else "N/A"}...
Final: {final_prompt[:500] if final_prompt else "N/A"}...
---
Generate skill content as JSON:
```json
{{
"title": "Human-readable skill title",
"description": "One-line description for skill discovery (what triggers this skill)",
"intro": "2-3 sentence introduction explaining what this skill teaches",
"activation": "Bullet points of when to activate this skill (specific keywords, task types)",
"concepts": "Core concepts this skill covers (3-5 key ideas)",
"anti_patterns": "Patterns to AVOID - formatted as markdown list with descriptions",
"practices": "Recommended practices - formatted as markdown list",
"guidelines": "Numbered verifiable guidelines (things that can be checked)",
"examples": "1-2 concrete before/after examples showing improvement"
}}
```"""
response = self.client.messages.create(
model=self.model,
max_tokens=4096,
system=GENERATOR_SYSTEM_PROMPT,
messages=[{"role": "user", "content": prompt}],
)
# Parse response
for block in response.content:
if block.type == "text":
try:
text = block.text
if "```json" in text:
text = text.split("```json")[1].split("```")[0]
return json.loads(text)
except json.JSONDecodeError:
pass
# Return defaults if parsing fails
return {
"title": "Generated Agent Skill",
"description": f"Optimized practices for {task}",
"intro": "This skill contains learnings from automated prompt optimization.",
"activation": "- When working on similar tasks\n- When debugging agent failures",
"concepts": "See recommendations section.",
"anti_patterns": patterns_text or "No patterns identified.",
"practices": recommendations_text or "No specific practices.",
"guidelines": "1. Review the anti-patterns before implementation\n2. Apply recommended practices",
"examples": "See optimization artifacts for detailed examples.",
}
def _save_references(
self,
skill_dir: Path,
result: LoopResult,
content: dict[str, str],
) -> None:
"""Save reference materials alongside the skill."""
refs_dir = skill_dir / "references"
refs_dir.mkdir(exist_ok=True)
# Save optimization summary
summary = {
"task": result.task,
"iterations": result.total_iterations,
"initial_score": result.initial_score,
"final_score": result.final_score,
"improvement": result.improvement_percentage,
"converged": result.converged,
"generated_at": datetime.now().isoformat(),
}
with open(refs_dir / "optimization_summary.json", "w") as f:
json.dump(summary, f, indent=2)
# Save final optimized prompt
with open(refs_dir / "optimized_prompt.txt", "w") as f:
f.write(result.final_prompt)
# Save all patterns found
patterns_data = []
for iteration in result.iterations:
for p in iteration.analysis.patterns:
patterns_data.append({
"type": p.type.value,
"severity": p.severity.value,
"description": p.description,
"suggestion": p.suggestion,
"iteration": iteration.iteration,
})
with open(refs_dir / "patterns_found.json", "w") as f:
json.dump(patterns_data, f, indent=2)
def generate_skill_from_loop(
result: LoopResult,
skill_name: str,
output_dir: str = "./generated_skills",
) -> str:
"""
Quick helper to generate a skill from optimization results.
Args:
result: Optimization loop result
skill_name: Name for the skill
output_dir: Output directory
Returns:
Path to generated skill
"""
generator = SkillGenerator()
return generator.generate(result, skill_name, output_dir)
tests/__init__.pyShowHide
"""Tests for Reasoning Trace Optimizer."""