context¶
第1章 · Agent 基础知识 · 配套项目
chapter1/context
项目说明¶
Context-Aware AI Agent with Ablation Studies¶
An advanced AI agent implementation supporting multiple LLM providers (SiliconFlow Qwen, ByteDance Doubao, Moonshot Kimi, and DeepSeek), designed to demonstrate the critical importance of context components through systematic ablation studies.
🎯 Overview¶
This project implements a context-aware AI agent with multiple tools (PDF parsing, currency conversion, calculator, code interpreter) and provides comprehensive ablation testing to explore how different context components affect agent behavior and performance.
Key Features¶
- Multi-provider Support: Works with SiliconFlow (Qwen), Doubao (ByteDance), Kimi (Moonshot), and DeepSeek LLMs
- Multi-tool Agent: PDF parsing, currency conversion, calculations, and Python code execution
- Context Modes: Five different context configurations for ablation studies
- Interactive & Batch Modes: Run single tasks or comprehensive test suites
- Conversation History: Maintains context across multiple queries in a session
- Detailed Analytics: Performance metrics, visualizations, and comprehensive reports
🤖 Supported LLM Providers¶
Doubao (ByteDance) - Default¶
- Model: doubao-seed-1-6-thinking-250715 (customizable)
- API: OpenAI-compatible via Volcano Engine
- Best for: Advanced reasoning, faster responses, both English and Chinese tasks
SiliconFlow¶
- Model: Qwen/Qwen3.5-397B-A17B (customizable)
- API: OpenAI-compatible
- Best for: Complex reasoning tasks, detailed analysis
Kimi (Moonshot AI)¶
- Model: kimi-k3 (K3 reasoning model; temperature is forced to 1 and max_tokens is large enough for its thinking output)
- API: OpenAI-compatible via Moonshot platform
- Best for: Advanced reasoning, multi-turn conversations, both English and Chinese tasks
- Features: Context caching for cost optimization
DeepSeek¶
- Model: deepseek-v4-flash (default; use
--model deepseek-v4-profor the stronger tier) - API: OpenAI-compatible via DeepSeek Platform
- Best for: Cost-effective tool-calling agents; thinking mode enabled so the
no_reasoningablation can stripreasoning_content - Note: Legacy aliases
deepseek-chat/deepseek-reasonerare deprecated (2026-07-24); prefer the V4 ids
🏗️ Architecture¶
Context Components¶
- Full Context - Complete agent with all components
- No History - Lacks historical tool call tracking
- No Reasoning - Operates without strategic planning
- No Tool Calls - Cannot execute external tools
- No Tool Results - Blind to tool execution outcomes
Available Tools¶
parse_pdf(url)- Download and extract text from PDF documentsconvert_currency(amount, from, to)- Real-time currency conversioncalculate(expression)- Simple mathematical expression evaluationcode_interpreter(code)- Execute Python code for complex calculations, totals, and data processing
📋 Prerequisites¶
- Python 3.8+
- API key for one of the supported providers:
- SiliconFlow: Get from SiliconFlow
- Doubao (ByteDance): Get from Volcano Engine
- Kimi (Moonshot): Get from Moonshot Platform
- DeepSeek: Get from DeepSeek Platform
📝 Sample Tasks¶
The system includes 5 pre-defined sample tasks demonstrating different capabilities:
- Simple Currency Conversion - Basic multi-currency calculations
- Multi-Currency Budget Analysis - Complex expense analysis across offices
- PDF Financial Analysis - Parse and analyze financial documents
- Investment Growth Calculation - Compound interest with currency conversion
- Comprehensive Financial Report - Complete workflow using all tools
These samples are designed to showcase the agent's capabilities and the impact of context ablation.
🚀 Quick Start¶
1. Installation¶
# Clone the repository
cd projects/week1/context
# Install dependencies
pip install -r requirements.txt
# Copy and configure environment
cp env.example .env
# Edit .env and add your API key (SILICONFLOW_API_KEY or ARK_API_KEY)
2. Configure Provider¶
# For Doubao (ByteDance) - Default
export ARK_API_KEY=your_key_here
python main.py # Uses Doubao by default
# For SiliconFlow (Qwen)
export SILICONFLOW_API_KEY=your_key_here
python main.py --provider siliconflow
# For Kimi (Moonshot)
export MOONSHOT_API_KEY=your_key_here
python main.py --provider kimi
# For DeepSeek
export DEEPSEEK_API_KEY=your_key_here
python main.py --provider deepseek
# Optional stronger model:
python main.py --provider deepseek --model deepseek-v4-pro
# Or specify a custom model
python main.py --model doubao-seed-1-6-thinking-250715
# Universal OpenRouter fallback: if the provider key above is missing/invalid
# but OPENROUTER_API_KEY is set, requests are routed through OpenRouter and the
# model id is mapped automatically (bare gpt-*/o1-* -> openai/*, claude-* ->
# anthropic/*, deepseek-* -> deepseek/*, other native ids -> OPENROUTER_MODEL
# or openai/gpt-5.6-luna).
export OPENROUTER_API_KEY=sk-or-v1-your-key-here
python main.py # falls back to OpenRouter when ARK_API_KEY is unset
python main.py --provider openrouter # or use OpenRouter directly
3. Testing Kimi / DeepSeek Integration¶
# Quick test of Kimi K3 model
export MOONSHOT_API_KEY=your_key_here
python test_kimi.py
# Use Kimi in main script
python main.py --provider kimi --mode interactive
# Run ablation study with Kimi
python main.py --provider kimi --mode ablation
# Quick test of DeepSeek V4
export DEEPSEEK_API_KEY=your_key_here
python test_deepseek.py
# or: python quick_test_deepseek.py
# Use DeepSeek in main script / ablation study
python main.py --provider deepseek --mode interactive
python main.py --provider deepseek --mode ablation
4. Run Interactive Mode (Recommended)¶
# Default (Doubao)
python main.py --mode interactive
# With SiliconFlow provider
python main.py --mode interactive --provider siliconflow
# In interactive mode, you can:
# - Type 'samples' to see pre-defined tasks
# - Type 'sample 3' to test PDF parsing
# - Type 'providers' to list available providers
# - Type 'provider kimi' to switch providers
# - Type 'status' to see current configuration
# - Type 'help' for all commands
5. Run Sample Tasks¶
# Run without arguments to select from samples
python main.py --mode single
# With specific provider
python main.py --mode single --provider doubao
# Or provide your own task
python main.py --mode single \
--task "Convert $1000 USD to EUR, GBP, and JPY. Calculate the average." \
--context-mode full \
--provider siliconflow
6. Run Ablation Study¶
# With default provider (single case, all five context modes)
python main.py --mode ablation
# With Doubao provider
python main.py --mode ablation --provider doubao
# Multi-case comparison across modes (stronger evidence for the book's point)
python main.py --mode ablation --cases 3
# Compare only two modes and save raw results to a custom path
python main.py --mode ablation --ablation-modes full no_history --output my_ablation.json
main.py is the single CLI entry point. Run python main.py --help for the
full (Chinese) flag reference.
Key flags:
| Flag | Description |
|---|---|
--mode |
single / ablation / interactive (default) |
--task |
Task text for single mode |
--context-mode |
Context mode for single mode (full, no_history, no_reasoning, no_tool_calls, no_tool_results) |
--ablation-modes |
Subset of modes to test in ablation mode (default: all five) |
--cases |
Number of cases each mode is run against in ablation mode (default: 1) |
--provider / --model |
LLM provider and optional model override |
--output |
Output path for the JSON result (single) or raw results (ablation) |
🧪 Ablation Studies¶
The ablation studies systematically remove context components to understand their importance:
Test Scenario¶
A complex financial analysis task requiring: 1. PDF document parsing 2. Multiple currency conversions 3. Mathematical calculations 4. Result aggregation
Expected Behaviors¶
| Context Mode | Removed Component (book §实验 1.1) | Expected Behavior | Impact |
|---|---|---|---|
| full | none (baseline) | Complete successful execution | Baseline performance |
| no_history | 历史消息 (history) | Redundant operations, inefficiency | May repeat tool calls |
| no_reasoning | 思考过程 (reasoning) | Unstructured approach, potential errors | Lacks strategic planning |
| no_tool_calls | 工具定义 (tool definitions) | Complete failure | Cannot interact with external world |
| no_tool_results | 工具执行结果 (tool results) | Incorrect conclusions | Makes decisions without feedback |
How each ablation is applied (see agent.py):
- no_tool_calls — the
toolsparameter is omitted from the request, so the model has no tool definitions to call. - no_tool_results — every tool result is replaced with a
[Tool result hidden]placeholder. - no_reasoning —
reasoning_contentis stripped from each assistant message before it is added back to the trajectory. - no_history —
_prepare_messages_for_api()sends only a sliding window (system prompt + current task + the most recent ReAct step) to the model, so earlier steps are forgotten and the agent tends to repeat tool calls. Full mode always sends the complete trajectory.
Running Tests¶
# Run the full ablation study (single case, all five modes)
python main.py --mode ablation
# Run across multiple cases for a stronger comparison
python main.py --mode ablation --cases 3
# This will generate:
# - ablation_study_results.png (visualization, if matplotlib is installed)
# - ablation_study_report.md (detailed report)
# - ablation_results.json (raw data; override path with --output)
The console prints two tables: a per-run ablation study results table and a comparison matrix (context mode x case) for reading the effect of each component at a glance.
📊 Understanding Results¶
Performance Metrics¶
- Success Rate: Whether the task was completed correctly
- Execution Time: Total time to complete the task
- Iterations: Number of agent-model interactions
- Tool Calls: Number of external tool invocations
- Reasoning Steps: Strategic planning iterations
Sample Output¶
ABLATION STUDY RESULTS
================================================================================
| Test Name | Success | Time | Iterations | Tool Calls |
|--------------------------------|---------|--------|------------|------------|
| Baseline - Full Context | ✓ | 12.3s | 5 | 8 |
| No Historical Tool Calls | ✓ | 18.7s | 8 | 12 |
| No Reasoning Process | ✗ | 25.4s | 10 | 15 |
| No Tool Call Commands | ✗ | 3.2s | 2 | 0 |
| No Tool Call Results | ✗ | 15.6s | 10 | 10 |
💡 Key Insights¶
1. Tool Calls Are Fundamental¶
Without tool call capability, the agent cannot interact with external systems, making task completion impossible.
2. Tool Results Provide Critical Feedback¶
Without seeing results, the agent operates blind, leading to incorrect conclusions and infinite loops.
3. Reasoning Enables Efficiency¶
Strategic planning reduces iterations and tool calls, improving both speed and accuracy.
4. History Prevents Redundancy¶
Historical context prevents repeated operations and maintains task coherence across iterations.
🛠️ Advanced Usage¶
Interactive Mode Commands¶
The interactive mode supports the following commands:
| Command | Description |
|---|---|
samples |
Display all available sample tasks |
sample <n> |
Run sample task number n |
providers |
List all available LLM providers |
provider <name> |
Switch to a different provider (e.g., provider kimi) |
modes |
List available context modes for ablation testing |
mode <name> |
Switch context mode (e.g., mode no_history) |
status |
Show current configuration (provider, model, mode, etc.) |
reset |
Reset agent trajectory (clear history) |
create_pdfs |
Generate sample PDF files for testing |
quit |
Exit interactive mode |
Note: The prompt shows the current provider in brackets, e.g., [KIMI]> or [DOUBAO]>
Conversation History¶
The agent maintains conversation history throughout interactive sessions:
- Persistent Context: The agent remembers previous queries and responses within a session
- Multi-turn Conversations: You can reference information from earlier in the conversation
- Tool Call Memory: Previous tool executions are remembered and can be referenced
- Reset on Demand: Use the
resetcommand to clear history and start fresh
Example conversation flow:
[DOUBAO]> Remember that our budget is $10,000. Calculate 15% of it.
# Agent calculates and remembers the budget
[DOUBAO]> Now convert that 15% amount to EUR
# Agent uses the previously calculated amount without re-asking
[DOUBAO]> What was our original budget?
# Agent recalls the $10,000 mentioned earlier
Custom Tasks¶
Create your own test scenarios:
from agent import ContextAwareAgent, ContextMode
agent = ContextAwareAgent(api_key, ContextMode.FULL)
result = agent.execute_task("""
Download the PDF from https://example.com/report.pdf,
extract all monetary values, convert them to EUR,
and calculate the total.
""")
Creating Test PDFs¶
Generate sample PDFs for testing:
Configuration¶
Edit config.py or set environment variables:
📁 Project Structure¶
context/
├── agent.py # Core agent implementation + context modes
├── main.py # Single CLI entry point (single / ablation / interactive)
├── config.py # Configuration management
├── create_sample_pdf.py # PDF generation utility
├── requirements.txt # Dependencies
├── env.example # Environment template
└── README.md # This file
Note: the ablation study lives in
main.py(AblationTestSuite), run viapython main.py --mode ablation. There is no separateablation_tests.py.
🔬 Research Applications¶
This implementation is valuable for:
- AI Safety Research: Understanding failure modes
- System Design: Identifying critical components
- Optimization: Finding minimal viable configurations
- Education: Teaching agent architecture principles
🤝 Contributing¶
Contributions are welcome! Areas for improvement:
- Additional tool implementations
- More sophisticated test scenarios
- Alternative context ablation strategies
- Performance optimizations
⚠️ Limitations¶
- Currency rates are fixed (production should use real-time APIs)
- PDF parsing may fail on complex layouts
- Model token limits may affect very large documents
📝 License¶
MIT License - See LICENSE file for details
🙏 Acknowledgments¶
- SiliconFlow for providing the Qwen model API
- DeepSeek for the OpenAI-compatible V4 API
- OpenAI for the client library
- The AI agent research community
📧 Contact¶
For questions or feedback, please open an issue on GitHub.
Note: This is an educational project demonstrating ablation studies in AI agents. For production use, implement proper error handling, rate limiting, and security measures.
源代码¶
agent.py¶
"""
Context-Aware AI Agent with Tool Calls
An agent using Qwen model from SiliconFlow with document parsing, currency conversion, and calculator tools.
Designed to demonstrate the importance of context through ablation studies.
"""
import json
import os
import re
import logging
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
import requests
from openai import OpenAI
import PyPDF2
from io import BytesIO
import math
from datetime import datetime
from concurrent.futures import TimeoutError
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
def _reasoning_safe_temperature(model, requested=1.0):
"""Reasoning models (Kimi K3, GPT-5, ...) only accept temperature=1.
Return 1 for those; otherwise the requested value so non-reasoning
providers (Doubao, DeepSeek, older Moonshot) are unchanged."""
m = str(model or "").lower().replace("/", "-")
return 1 if ("kimi-k3" in m or "gpt-5" in m) else requested
class ContextMode(Enum):
"""Different context modes for ablation studies"""
FULL = "full" # Complete context with all components
NO_HISTORY = "no_history" # No historical tool calls
NO_REASONING = "no_reasoning" # No reasoning/thinking process
NO_TOOL_CALLS = "no_tool_calls" # No tool call commands
NO_TOOL_RESULTS = "no_tool_results" # No tool call results
@dataclass
class ToolCall:
"""Represents a single tool call"""
tool_name: str
arguments: Dict[str, Any]
result: Optional[Any] = None
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
@dataclass
class AgentTrajectory:
"""Tracks the agent's execution trajectory"""
reasoning_steps: List[str] = field(default_factory=list)
tool_calls: List[ToolCall] = field(default_factory=list)
context_mode: ContextMode = ContextMode.FULL
class ToolRegistry:
"""Registry for available tools"""
@staticmethod
def parse_pdf(url: str) -> Dict[str, Any]:
"""
Download and parse a PDF from URL or local file
Args:
url: URL or file path of the PDF to parse
Returns:
Dictionary containing parsed text and metadata
"""
try:
# Check if it's a local file
if url.startswith('file://'):
# Extract the file path from file:// URL
file_path = url.replace('file://', '')
logger.info(f"Reading local PDF from {file_path}")
# Read the file directly
with open(file_path, 'rb') as f:
pdf_content = f.read()
elif url.startswith('/') or url.startswith('./') or url.startswith('../') or ':\\' in url or ':/' in url[1:3]:
# Direct file path (absolute or relative)
logger.info(f"Reading local PDF from {url}")
# Read the file directly
with open(url, 'rb') as f:
pdf_content = f.read()
else:
# It's a remote URL, download it
logger.info(f"Downloading PDF from {url}")
response = requests.get(url, timeout=30)
response.raise_for_status()
pdf_content = response.content
# Parse the PDF content
pdf_file = BytesIO(pdf_content)
pdf_reader = PyPDF2.PdfReader(pdf_file)
text_content = []
for page_num, page in enumerate(pdf_reader.pages, 1):
text = page.extract_text()
text_content.append({
"page": page_num,
"text": text
})
result = {
"url": url,
"num_pages": len(pdf_reader.pages),
"content": text_content,
"metadata": pdf_reader.metadata if hasattr(pdf_reader, 'metadata') else {}
}
logger.info(f"Successfully parsed PDF with {len(pdf_reader.pages)} pages")
return result
except Exception as e:
logger.error(f"Error parsing PDF: {str(e)}")
return {"error": str(e)}
@staticmethod
def convert_currency(amount: float, from_currency: str, to_currency: str) -> Dict[str, Any]:
"""
Convert currency using live exchange rates
Args:
amount: Amount to convert
from_currency: Source currency code (e.g., 'USD')
to_currency: Target currency code (e.g., 'EUR')
Returns:
Dictionary with conversion result
"""
try:
# Normalize currency codes (handle S$ notation)
from_currency = from_currency.replace("S$", "SGD").replace("$", "USD") if from_currency.startswith("S$") else from_currency
to_currency = to_currency.replace("S$", "SGD").replace("$", "USD") if to_currency.startswith("S$") else to_currency
logger.info(f"Converting {amount} {from_currency} to {to_currency}")
# For demonstration, using fixed rates (in production, use a real API)
# These are example rates - you would normally fetch from an API
exchange_rates = {
"USD": 1.0,
"EUR": 0.92,
"GBP": 0.79,
"JPY": 149.50,
"CNY": 7.24,
"CAD": 1.36,
"AUD": 1.53,
"CHF": 0.88,
"INR": 83.12,
"SGD": 1.34
}
if from_currency not in exchange_rates or to_currency not in exchange_rates:
return {"error": f"Unsupported currency: {from_currency} or {to_currency}"}
# Convert to USD first, then to target currency
usd_amount = amount / exchange_rates[from_currency]
converted_amount = usd_amount * exchange_rates[to_currency]
result = {
"original_amount": amount,
"from_currency": from_currency,
"to_currency": to_currency,
"converted_amount": round(converted_amount, 2),
"exchange_rate": round(exchange_rates[to_currency] / exchange_rates[from_currency], 4),
"timestamp": datetime.now().isoformat()
}
logger.info(f"Conversion result: {result['converted_amount']} {to_currency}")
return result
except Exception as e:
logger.error(f"Error converting currency: {str(e)}")
return {"error": str(e)}
@staticmethod
def calculate(expression: str) -> Dict[str, Any]:
"""
Evaluate a mathematical expression
Args:
expression: Mathematical expression to evaluate
Returns:
Dictionary with calculation result
"""
try:
logger.info(f"Calculating: {expression}")
# Sanitize expression - only allow safe mathematical operations
allowed_names = {
k: v for k, v in math.__dict__.items() if not k.startswith("__")
}
allowed_names.update({"abs": abs, "round": round, "min": min, "max": max})
# Replace common operations for clarity
expression = expression.replace("^", "**")
# Evaluate the expression
result = eval(expression, {"__builtins__": {}}, allowed_names)
return {
"expression": expression,
"result": result,
"type": type(result).__name__
}
except Exception as e:
logger.error(f"Error calculating expression: {str(e)}")
return {"error": str(e)}
@staticmethod
def code_interpreter(code: str) -> Dict[str, Any]:
"""
Execute Python code for complex calculations and data processing
Args:
code: Python code to execute
Returns:
Dictionary with execution results and any output
"""
try:
logger.info(f"Executing Python code: {code[:100]}...")
# Create a restricted namespace with safe built-ins
safe_namespace = {
'__builtins__': {
'abs': abs,
'all': all,
'any': any,
'sum': sum,
'min': min,
'max': max,
'round': round,
'len': len,
'list': list,
'dict': dict,
'set': set,
'tuple': tuple,
'enumerate': enumerate,
'zip': zip,
'map': map,
'filter': filter,
'sorted': sorted,
'reversed': reversed,
'range': range,
'int': int,
'float': float,
'str': str,
'bool': bool,
'print': print,
}
}
# Add math module
safe_namespace['math'] = math
# Capture printed output
import io
import contextlib
output_buffer = io.StringIO()
with contextlib.redirect_stdout(output_buffer):
# Execute the code
exec(code, safe_namespace)
# Get printed output
printed_output = output_buffer.getvalue()
# Try to extract a result if it's assigned to 'result' variable
result = safe_namespace.get('result', None)
# Also check for common variable names
if result is None:
for var_name in ['total', 'sum', 'output', 'answer', 'final']:
if var_name in safe_namespace:
result = safe_namespace[var_name]
break
# Get all variables defined (excluding built-ins and modules)
variables = {
k: v for k, v in safe_namespace.items()
if not k.startswith('__') and k not in ['math'] and not callable(v)
}
return {
"code": code,
"result": result,
"output": printed_output if printed_output else None,
"variables": variables,
"success": True
}
except Exception as e:
logger.error(f"Error executing code: {str(e)}")
return {
"code": code,
"error": str(e),
"success": False
}
class ContextAwareAgent:
"""
AI Agent with configurable LLM providers and context modes for ablation studies
"""
def __init__(self, api_key: str, context_mode: ContextMode = ContextMode.FULL,
provider: str = "siliconflow", model: Optional[str] = None,
verbose: bool = True):
"""
Initialize the agent
Args:
api_key: API key for the LLM provider
context_mode: Context mode for ablation studies
provider: LLM provider ('siliconflow', 'doubao', 'kimi', 'moonshot',
'deepseek', or 'openrouter')
model: Optional model override
verbose: If True, log full HTTP requests and responses (default: True)
"""
self.provider = provider.lower()
self.verbose = verbose
# Provider -> (base_url, default_model)
deepseek_base = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
provider_defaults = {
"siliconflow": ("https://api.siliconflow.cn/v1", "Qwen/Qwen3.5-397B-A17B"),
"doubao": ("https://ark.cn-beijing.volces.com/api/v3", "doubao-seed-1-6-thinking-250715"),
"kimi": ("https://api.moonshot.cn/v1", "kimi-k3"),
"moonshot": ("https://api.moonshot.cn/v1", "kimi-k3"),
# V4 Flash: OpenAI-compatible; tool calling + thinking mode.
# Legacy deepseek-chat / deepseek-reasoner aliases deprecated 2026-07-24.
"deepseek": (deepseek_base, "deepseek-v4-flash"),
"openrouter": ("https://openrouter.ai/api/v1", "openai/gpt-5.6-luna"),
}
if self.provider not in provider_defaults:
raise ValueError(
f"Unsupported provider: {provider}. Use 'siliconflow', 'doubao', "
"'kimi', 'moonshot', 'deepseek', or 'openrouter'"
)
base_url, default_model = provider_defaults[self.provider]
resolved_model = model or default_model
# Universal OpenRouter fallback: if the primary provider key is missing
# but OPENROUTER_API_KEY is present, route through OpenRouter with a
# mapped model id. Behavior is unchanged when the provider key is set.
from config import resolve_llm_backend
resolved_key, resolved_base_url, self.model, self.using_openrouter = \
resolve_llm_backend(api_key, base_url, resolved_model)
if self.using_openrouter:
logger.info(
f"{self.provider} API key not set; routing via OpenRouter "
f"(model: {self.model})"
)
self.client = OpenAI(
api_key=resolved_key,
base_url=resolved_base_url
)
self.context_mode = context_mode
self.trajectory = AgentTrajectory(context_mode=context_mode)
self.tools = ToolRegistry()
# Initialize conversation history
self.conversation_history = []
self._init_system_prompt()
logger.info(f"Agent initialized with provider: {self.provider}, model: {self.model}, context mode: {context_mode.value}, verbose: {self.verbose}")
def _init_system_prompt(self):
"""Initialize the system prompt for the conversation"""
self.conversation_history = [
{
"role": "system",
"content": """You are an intelligent assistant with access to tools.
Your task is to solve the given problems using the available tools. Think step by step and use tools as needed.
Important: When you have gathered all necessary information and computed the final answer, clearly state "FINAL ANSWER:" followed by your answer."""
}
]
def _get_tools_description(self) -> List[Dict[str, Any]]:
"""Get tool descriptions for the model"""
return [
{
"type": "function",
"function": {
"name": "parse_pdf",
"description": "Download and parse a PDF document from a URL to extract text content",
"parameters": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "The URL of the PDF document to parse"
}
},
"required": ["url"]
}
}
},
{
"type": "function",
"function": {
"name": "convert_currency",
"description": "Convert an amount from one currency to another using current exchange rates",
"parameters": {
"type": "object",
"properties": {
"amount": {
"type": "number",
"description": "The amount to convert"
},
"from_currency": {
"type": "string",
"description": "The source currency code (e.g., USD, EUR)"
},
"to_currency": {
"type": "string",
"description": "The target currency code (e.g., USD, EUR)"
}
},
"required": ["amount", "from_currency", "to_currency"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "Evaluate a simple mathematical expression",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "The mathematical expression to evaluate (e.g., '2 + 2 * 3')"
}
},
"required": ["expression"]
}
}
},
{
"type": "function",
"function": {
"name": "code_interpreter",
"description": "Execute Python code for complex calculations, data processing, and computing totals. Use this for tasks like: summing lists of values, calculating percentages, aggregating financial data, performing multi-step calculations, or any computation requiring variables and intermediate steps.",
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "Python code to execute. Can use variables, loops, and mathematical operations. Example: 'amounts = [2500000, 2278481, 2541806, 2282609, 2388060]; total = sum(amounts); print(f\"Total: ${total:,.2f}\")"
}
},
"required": ["code"]
}
}
}
]
def _prepare_assistant_message(self, message) -> Dict[str, Any]:
"""
Prepare assistant message for adding to messages list,
filtering out reasoning_content if in NO_REASONING mode
Args:
message: The assistant message object
Returns:
Dictionary representation of the message
"""
msg_dict = message.dict() if hasattr(message, 'dict') else message.model_dump()
# Remove reasoning_content if in NO_REASONING mode
if self.context_mode == ContextMode.NO_REASONING and 'reasoning_content' in msg_dict:
msg_dict.pop('reasoning_content')
return msg_dict
def _build_context(self) -> str:
"""
Build a human-readable summary of the trajectory (legacy helper, kept
for inspection/debugging only).
NOTE: The message list sent to the model is assembled by
``_prepare_messages_for_api`` -- that is where the NO_HISTORY ablation
actually takes effect. This method is not part of the request path.
Returns:
Context string for the model
"""
context_parts = []
# Add reasoning steps if not disabled
if self.context_mode != ContextMode.NO_REASONING and self.trajectory.reasoning_steps:
context_parts.append("## Previous Reasoning Steps:")
for step in self.trajectory.reasoning_steps:
context_parts.append(f"- {step}")
context_parts.append("")
# Add tool call history if not disabled
if self.context_mode not in [ContextMode.NO_HISTORY, ContextMode.NO_TOOL_CALLS] and self.trajectory.tool_calls:
context_parts.append("## Tool Call History:")
for call in self.trajectory.tool_calls:
if self.context_mode != ContextMode.NO_TOOL_CALLS:
context_parts.append(f"- Called {call.tool_name} with args: {json.dumps(call.arguments)}")
if self.context_mode != ContextMode.NO_TOOL_RESULTS and call.result:
context_parts.append(f" Result: {json.dumps(call.result, indent=2)}")
context_parts.append("")
return "\n".join(context_parts) if context_parts else ""
def _log_request_response(self, request_data: Dict[str, Any], response_data: Any, iteration: int):
"""
Log full request and response when in verbose mode
Args:
request_data: The request payload sent to the API
response_data: The response received from the API
iteration: Current iteration number
"""
if not self.verbose:
return
if request_data:
print("\n" + "="*80)
print(f"📤 ITERATION {iteration} - FULL REQUEST JSON:")
print("-"*80)
print(json.dumps(request_data, indent=2, ensure_ascii=False))
if response_data:
print("\n" + "="*80)
print(f"📥 ITERATION {iteration} - FULL RESPONSE:")
print("-"*80)
# Convert response to dict for display
if hasattr(response_data, 'model_dump'):
response_dict = response_data.model_dump()
elif hasattr(response_data, 'dict'):
response_dict = response_data.dict()
else:
response_dict = {"raw_response": str(response_data)}
print(json.dumps(response_dict, indent=2, ensure_ascii=False))
print("="*80 + "\n")
def _execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""
Execute a tool and return the result
Args:
tool_name: Name of the tool to execute
arguments: Arguments for the tool
Returns:
Tool execution result
"""
tool_map = {
"parse_pdf": self.tools.parse_pdf,
"convert_currency": self.tools.convert_currency,
"calculate": self.tools.calculate,
"code_interpreter": self.tools.code_interpreter
}
if tool_name not in tool_map:
return {"error": f"Unknown tool: {tool_name}"}
return tool_map[tool_name](**arguments)
def _prepare_messages_for_api(self) -> List[Dict[str, Any]]:
"""
Build the message list actually sent to the model for the current
iteration, applying the NO_HISTORY ablation.
For every mode except NO_HISTORY the full conversation history (the
accumulated trajectory) is returned unchanged. For NO_HISTORY a sliding
window is returned that keeps only:
- the system prompt (static prefix), and
- the latest user task plus the MOST RECENT ReAct step (the last
assistant message together with the tool results that follow it).
All earlier steps are dropped, so the agent "forgets" what it already
did and tends to repeat tool calls -- exactly the failure mode the book
attributes to missing 历史消息 (history). Keeping the last assistant
message together with its trailing tool messages preserves API validity
(tool results stay paired with their assistant tool_calls).
Returns:
The message list to send to the model for this iteration.
"""
messages = self.conversation_history
if self.context_mode != ContextMode.NO_HISTORY:
return messages
# System prompt(s) are always kept as the static prefix.
windowed = [m for m in messages if m.get("role") == "system"]
# Anchor on the latest user task.
user_indices = [i for i, m in enumerate(messages) if m.get("role") == "user"]
if not user_indices:
return windowed
last_user_idx = user_indices[-1]
windowed.append(messages[last_user_idx])
# Keep only the most recent step after the task: from the last assistant
# message to the end. Its tool results follow it, so the pairing stays
# valid while every earlier step is dropped.
tail = messages[last_user_idx + 1:]
assistant_rel = [i for i, m in enumerate(tail) if m.get("role") == "assistant"]
if assistant_rel:
windowed.extend(tail[assistant_rel[-1]:])
return windowed
@staticmethod
def _extract_final_answer(content: str) -> Optional[str]:
"""Extract text after FINAL ANSWER: if present; otherwise None."""
if not content or "FINAL ANSWER:" not in content:
return None
return content.split("FINAL ANSWER:", 1)[1].strip()
def execute_task(self, task: str, max_iterations: Optional[int] = None) -> Dict[str, Any]:
"""
Execute a task using available tools (ReAct loop).
Stops when:
1. The model emits a text-only reply (no tool_calls) — conversational
or task complete, including plain replies like "hi" that omit the
FINAL ANSWER: marker; or
2. max_iterations is hit (safety cap for tool-call loops, e.g. the
no_tool_results ablation).
Args:
task: The task to execute
max_iterations: Maximum ReAct steps (default: Config.MAX_ITERATIONS
or 10). This is a safety ceiling, not a target round count.
Returns:
Task execution result
"""
if max_iterations is None:
try:
from config import Config
max_iterations = Config.MAX_ITERATIONS
except Exception:
max_iterations = 10
# Add user message to conversation history
self.conversation_history.append({"role": "user", "content": task})
# Use conversation history directly (no copy needed)
messages = self.conversation_history
iteration = 0
final_answer = None
while iteration < max_iterations:
iteration += 1
logger.info(f"Iteration {iteration}/{max_iterations}")
try:
# Build the message list actually sent to the model. For every
# mode except NO_HISTORY this equals the full trajectory; for
# NO_HISTORY it is a sliding window that drops earlier steps.
api_messages = self._prepare_messages_for_api()
# Prepare request data for logging
request_data = {
"model": self.model,
"messages": api_messages,
"temperature": _reasoning_safe_temperature(self.model, 0.3),
"max_tokens": 8192
}
if self.context_mode != ContextMode.NO_TOOL_CALLS:
request_data["tools"] = self._get_tools_description()
request_data["tool_choice"] = "auto"
# DeepSeek V4: enable thinking so reasoning_content is present
# for the no_reasoning ablation (parity with thinking defaults of
# Doubao/Kimi). Skip when routed via OpenRouter, which may not
# accept the same extra body shape.
create_kwargs = {
"model": self.model,
"messages": api_messages,
"tools": self._get_tools_description() if self.context_mode != ContextMode.NO_TOOL_CALLS else None,
"tool_choice": "auto" if self.context_mode != ContextMode.NO_TOOL_CALLS else None,
"temperature": _reasoning_safe_temperature(self.model, 0.3),
"max_tokens": 8192,
"timeout": 180, # 180 second timeout for main execution
}
if self.provider == "deepseek" and not getattr(self, "using_openrouter", False):
create_kwargs["extra_body"] = {"thinking": {"type": "enabled"}}
request_data["thinking"] = {"type": "enabled"}
logger.info(f"Sending request to {self.provider} API")
# Call the model with tools
response = self.client.chat.completions.create(**create_kwargs)
# Log response if verbose
if self.verbose:
self._log_request_response(request_data, response, iteration)
message = response.choices[0].message
has_tool_calls = bool(getattr(message, "tool_calls", None))
# --- Terminal path: text reply with no tool calls ---
# A normal chat turn ("hi" -> "Hello!") or a task answer without
# the FINAL ANSWER: marker must end the ReAct loop. Previously
# only "FINAL ANSWER:" broke the loop, so plain replies were
# re-sent for up to max_iterations (wasted API calls).
if not has_tool_calls:
assistant_msg = self._prepare_assistant_message(message)
messages.append(assistant_msg)
content = (message.content or "").strip()
if content:
marked = self._extract_final_answer(content)
final_answer = marked if marked is not None else content
logger.info(
"Terminal text response (no tool calls); "
f"stopping after iteration {iteration}"
)
else:
logger.warning(
"Empty model response with no tool calls; "
"stopping to avoid burning remaining iterations"
)
break
# --- Continue path: model requested tool execution ---
assistant_msg = self._prepare_assistant_message(message)
messages.append(assistant_msg)
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
logger.info(f"Executing tool: {function_name} with args: {function_args}")
result = self._execute_tool(function_name, function_args)
tool_call_record = ToolCall(
tool_name=function_name,
arguments=function_args,
result=result
)
self.trajectory.tool_calls.append(tool_call_record)
if self.context_mode != ContextMode.NO_TOOL_RESULTS:
tool_msg = {
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
}
else:
tool_msg = {
"role": "tool",
"tool_call_id": tool_call.id,
"content": "[Tool result hidden due to context mode]"
}
messages.append(tool_msg)
# If the same turn also tagged FINAL ANSWER: (unusual with tools),
# still prefer extracting it after tools are recorded.
if message.content and "FINAL ANSWER:" in message.content:
final_answer = self._extract_final_answer(message.content)
logger.info(f"Final answer found alongside tool calls: {final_answer}")
break
# Note: We do NOT modify the system prompt anymore.
# The context is already built into the conversation through tool history
except TimeoutError as e:
logger.error(f"Request timed out after 60 seconds")
return {
"error": "Request timed out. The model is taking too long to respond. Try a simpler task or different provider.",
"trajectory": self.trajectory,
"iterations": iteration
}
except Exception as e:
logger.error(f"Error during task execution: {str(e)}")
# Check if it's a timeout-related error
if "timeout" in str(e).lower() or "timed out" in str(e).lower():
return {
"error": "Request timed out. The model is taking too long to respond. Try a simpler task or different provider.",
"trajectory": self.trajectory,
"iterations": iteration
}
return {
"error": str(e),
"trajectory": self.trajectory,
"iterations": iteration
}
return {
"final_answer": final_answer,
"trajectory": self.trajectory,
"iterations": iteration,
"success": final_answer is not None
}
def reset(self):
"""Reset the agent's trajectory and conversation history"""
self.trajectory = AgentTrajectory(context_mode=self.context_mode)
self._init_system_prompt() # Reinitialize conversation with system prompt
logger.info("Agent trajectory and conversation history reset")
def process(self, query: str, max_iterations: Optional[int] = None) -> str:
"""
Process a query and return the final answer as a string
Args:
query: The query to process
max_iterations: Maximum ReAct steps (default from Config)
Returns:
The final answer as a string
"""
result = self.execute_task(query, max_iterations)
if result.get('final_answer'):
return result['final_answer']
elif result.get('error'):
return f"Error: {result['error']}"
else:
return "No answer found"
config.py¶
"""
Configuration module for Context-Aware Agent
"""
import os
from typing import Optional
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
def _reasoning_safe_temperature(model, requested=1.0):
"""Reasoning models (Kimi K3, GPT-5, ...) only accept temperature=1.
Return 1 for those; otherwise the requested value so non-reasoning
providers (Doubao, DeepSeek, older Moonshot) are unchanged."""
m = str(model or "").lower().replace("/", "-")
return 1 if ("kimi-k3" in m or "gpt-5" in m) else requested
def map_model_to_openrouter(model: str) -> str:
"""Map a bare model id to an OpenRouter model id.
- ids already containing '/' -> left as-is
- gpt-*/o1-*/o3-*/o4-* -> 'openai/<id>'
- claude-* -> anthropic Claude (opus/sonnet/haiku)
- deepseek-* -> deepseek/<id> (OpenRouter hosts official DeepSeek ids)
- other native ids (kimi-*, doubao-*, ...) are NOT reliably on OpenRouter,
so fall back to OPENROUTER_MODEL or a safe default that always works.
"""
m = (model or "").strip()
if "/" in m:
return m
ml = m.lower()
if ml.startswith(("gpt-", "o1-", "o3-", "o4-")):
return "openai/" + m
if ml.startswith("claude-"):
if "sonnet" in ml:
return "anthropic/claude-sonnet-4.6"
if "haiku" in ml:
return "anthropic/claude-haiku-4.5"
return "anthropic/claude-opus-4.8"
if ml.startswith("kimi"):
# kimi-k3 is not on OpenRouter; moonshotai/kimi-k2.6 is the closest hosted id.
return "moonshotai/kimi-k2.6"
if ml.startswith("deepseek"):
# OpenRouter hosts deepseek/deepseek-v4-flash, deepseek-chat, etc.
return "deepseek/" + m
return os.getenv("OPENROUTER_MODEL", "openai/gpt-5.6-luna")
def resolve_llm_backend(primary_key: str, primary_base_url: str, model: str):
"""Universal OpenRouter fallback for LLM backend resolution.
Returns (api_key, base_url, model, using_openrouter).
- If the primary provider key is present, behavior is unchanged.
- Else if OPENROUTER_API_KEY is present, route through OpenRouter and map
the model id to an OpenRouter id.
- Else raise a clear error listing the accepted keys.
"""
openrouter_key = os.getenv("OPENROUTER_API_KEY")
# gpt-5.x (incl. gpt-5.6*) needs OpenAI org-verification on the direct API;
# when an OpenRouter key is present, prefer routing these ids through it.
if openrouter_key and str(model or "").lower().startswith("gpt-5"):
base_url = os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1")
return openrouter_key, base_url, map_model_to_openrouter(model), True
if primary_key:
return primary_key, primary_base_url, model, False
if openrouter_key:
base_url = os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1")
return openrouter_key, base_url, map_model_to_openrouter(model), True
raise ValueError(
"No API key found. Set a provider key "
"(SILICONFLOW_API_KEY/ARK_API_KEY/MOONSHOT_API_KEY/DEEPSEEK_API_KEY) or "
"OPENROUTER_API_KEY (universal fallback)."
)
class Config:
"""Configuration settings for the agent"""
# Provider Configuration
LLM_PROVIDER: str = os.getenv("LLM_PROVIDER", "doubao").lower()
# API Configuration
SILICONFLOW_API_KEY: str = os.getenv("SILICONFLOW_API_KEY", "")
SILICONFLOW_BASE_URL: str = "https://api.siliconflow.cn/v1"
ARK_API_KEY: str = os.getenv("ARK_API_KEY", "")
ARK_BASE_URL: str = "https://ark.cn-beijing.volces.com/api/v3"
MOONSHOT_API_KEY: str = os.getenv("MOONSHOT_API_KEY", "")
MOONSHOT_BASE_URL: str = "https://api.moonshot.cn/v1"
DEEPSEEK_API_KEY: str = os.getenv("DEEPSEEK_API_KEY", "")
DEEPSEEK_BASE_URL: str = os.getenv(
"DEEPSEEK_BASE_URL", "https://api.deepseek.com"
)
# Model Configuration (defaults based on provider)
MODEL_NAME: str = os.getenv("MODEL_NAME", "") # Will be set based on provider if not specified
MODEL_TEMPERATURE: float = float(os.getenv("MODEL_TEMPERATURE", "0.3"))
MODEL_MAX_TOKENS: int = int(os.getenv("MODEL_MAX_TOKENS", "1000"))
# Agent Configuration
MAX_ITERATIONS: int = int(os.getenv("MAX_ITERATIONS", "10"))
ENABLE_REASONING: bool = os.getenv("ENABLE_REASONING", "true").lower() == "true"
# Test Configuration
TEST_PDF_URL: str = os.getenv(
"TEST_PDF_URL",
"https://www.berkshirehathaway.com/qtrly/1stqtr23.pdf"
)
# Currency Configuration (Example rates - in production use real API)
EXCHANGE_RATES = {
"USD": 1.0,
"EUR": 0.92,
"GBP": 0.79,
"JPY": 149.50,
"CNY": 7.24,
"CAD": 1.36,
"AUD": 1.53,
"CHF": 0.88,
"INR": 83.12,
"SGD": 1.34
}
# Logging Configuration
LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")
LOG_FILE: Optional[str] = os.getenv("LOG_FILE")
LOG_FORMAT: str = "%(asctime)s - %(levelname)s - %(name)s - %(message)s"
# File paths
RESULTS_DIR: str = "results"
TEST_PDFS_DIR: str = "test_pdfs"
@classmethod
def get_api_key(cls, provider: str = None) -> str:
"""
Get API key for the specified provider
Args:
provider: Provider name (defaults to LLM_PROVIDER)
Returns:
API key for the provider
"""
provider = provider or cls.LLM_PROVIDER
provider = provider.lower()
if provider == "siliconflow":
return cls.SILICONFLOW_API_KEY
elif provider == "doubao":
return cls.ARK_API_KEY
elif provider == "kimi" or provider == "moonshot":
return cls.MOONSHOT_API_KEY
elif provider == "deepseek":
return cls.DEEPSEEK_API_KEY
else:
return ""
@classmethod
def get_default_model(cls, provider: str = None) -> str:
"""
Get default model for the specified provider
Args:
provider: Provider name (defaults to LLM_PROVIDER)
Returns:
Default model name for the provider
"""
provider = provider or cls.LLM_PROVIDER
provider = provider.lower()
if cls.MODEL_NAME:
return cls.MODEL_NAME
if provider == "siliconflow":
return "Qwen/Qwen3.5-397B-A17B"
elif provider == "doubao":
return "doubao-seed-1-6-thinking-250715"
elif provider == "kimi" or provider == "moonshot":
return "kimi-k3"
elif provider == "deepseek":
# V4 Flash: tool calling + thinking mode (legacy deepseek-chat /
# deepseek-reasoner aliases are deprecated 2026-07-24).
return "deepseek-v4-flash"
else:
return ""
@classmethod
def validate(cls, provider: str = None) -> bool:
"""
Validate required configuration
Args:
provider: Provider to validate (defaults to LLM_PROVIDER)
Returns:
True if configuration is valid
"""
provider = provider or cls.LLM_PROVIDER
api_key = cls.get_api_key(provider)
if not api_key:
if provider == "siliconflow":
print("ERROR: SILICONFLOW_API_KEY is not set")
elif provider == "doubao":
print("ERROR: ARK_API_KEY is not set")
elif provider == "kimi" or provider == "moonshot":
print("ERROR: MOONSHOT_API_KEY is not set")
elif provider == "deepseek":
print("ERROR: DEEPSEEK_API_KEY is not set")
else:
print(f"ERROR: No API key configured for provider: {provider}")
print("Please set it in .env file or as environment variable")
return False
return True
@classmethod
def create_directories(cls):
"""Create necessary directories if they don't exist"""
os.makedirs(cls.RESULTS_DIR, exist_ok=True)
os.makedirs(cls.TEST_PDFS_DIR, exist_ok=True)
@classmethod
def get_model_config(cls) -> dict:
"""
Get model configuration as dictionary
Returns:
Model configuration dict
"""
return {
"model": cls.MODEL_NAME,
"temperature": _reasoning_safe_temperature(cls.MODEL_NAME, cls.MODEL_TEMPERATURE),
"max_tokens": cls.MODEL_MAX_TOKENS
}
@classmethod
def print_config(cls):
"""Print current configuration (hiding sensitive data)"""
print("\n" + "="*50)
print("CONFIGURATION")
print("="*50)
print(f"Model: {cls.MODEL_NAME}")
print(f"Temperature: {cls.MODEL_TEMPERATURE}")
print(f"Max Tokens: {cls.MODEL_MAX_TOKENS}")
print(f"Max Iterations: {cls.MAX_ITERATIONS}")
print(f"API Key Set: {'Yes' if cls.SILICONFLOW_API_KEY else 'No'}")
print(f"Log Level: {cls.LOG_LEVEL}")
print("="*50 + "\n")
create_sample_pdf.py¶
"""
Create Sample PDF for Testing
Generates a financial report PDF with various currency amounts and calculations
"""
from reportlab.lib import colors
from reportlab.lib.pagesizes import letter, A4
from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer, PageBreak
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import inch
from reportlab.lib.enums import TA_CENTER, TA_RIGHT
import os
def create_financial_report():
"""Create a sample financial report PDF for testing"""
# Create PDF
filename = "sample_financial_report_q1_2024.pdf"
doc = SimpleDocTemplate(filename, pagesize=letter)
# Container for the 'Flowable' objects
elements = []
# Define styles
styles = getSampleStyleSheet()
title_style = ParagraphStyle(
'CustomTitle',
parent=styles['Heading1'],
fontSize=24,
textColor=colors.HexColor('#1f4788'),
spaceAfter=30,
alignment=TA_CENTER
)
heading_style = ParagraphStyle(
'CustomHeading',
parent=styles['Heading2'],
fontSize=16,
textColor=colors.HexColor('#1f4788'),
spaceAfter=12,
)
# Title
elements.append(Paragraph("Global Corporation Financial Report", title_style))
elements.append(Paragraph("Q1 2024 - Quarterly Results", styles['Heading2']))
elements.append(Spacer(1, 0.5*inch))
# Executive Summary
elements.append(Paragraph("Executive Summary", heading_style))
summary_text = """This report presents the financial performance of Global Corporation
for the first quarter of 2024. The company operates in multiple regions with
transactions in various currencies. Total consolidated revenue for Q1 2024
reached $45.8 million USD, representing a 12% increase year-over-year."""
elements.append(Paragraph(summary_text, styles['BodyText']))
elements.append(Spacer(1, 0.3*inch))
# Regional Revenue Table
elements.append(Paragraph("Regional Revenue Breakdown", heading_style))
revenue_data = [
['Region', 'Local Currency', 'Q1 2024 Revenue', 'Q4 2023 Revenue', 'Growth %'],
['North America', 'USD', '$15,250,000', '$14,100,000', '8.16%'],
['Europe', 'EUR', '€11,340,000', '€10,800,000', '5.00%'],
['United Kingdom', 'GBP', '£8,920,000', '£8,500,000', '4.94%'],
['Asia Pacific', 'JPY', '¥1,245,000,000', '¥1,180,000,000', '5.51%'],
['Singapore', 'SGD', 'S$4,180,000', 'S$3,950,000', '5.82%'],
]
revenue_table = Table(revenue_data, colWidths=[2*inch, 1.2*inch, 1.5*inch, 1.5*inch, 0.8*inch])
revenue_table.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, 0), colors.grey),
('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke),
('ALIGN', (0, 0), (-1, -1), 'CENTER'),
('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
('FONTSIZE', (0, 0), (-1, 0), 12),
('BOTTOMPADDING', (0, 0), (-1, 0), 12),
('BACKGROUND', (0, 1), (-1, -1), colors.beige),
('GRID', (0, 0), (-1, -1), 1, colors.black),
]))
elements.append(revenue_table)
elements.append(Spacer(1, 0.3*inch))
# Operating Expenses
elements.append(Paragraph("Operating Expenses by Department", heading_style))
expense_data = [
['Department', 'Q1 2024 (USD)', 'Q4 2023 (USD)', 'Change'],
['Research & Development', '$8,450,000', '$7,900,000', '+$550,000'],
['Sales & Marketing', '$6,230,000', '$6,100,000', '+$130,000'],
['General & Administrative', '$4,180,000', '$4,050,000', '+$130,000'],
['Operations', '$9,870,000', '$9,500,000', '+$370,000'],
['Total Operating Expenses', '$28,730,000', '$27,550,000', '+$1,180,000'],
]
expense_table = Table(expense_data, colWidths=[2.5*inch, 1.5*inch, 1.5*inch, 1.2*inch])
expense_table.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, 0), colors.grey),
('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke),
('ALIGN', (0, 0), (-1, -1), 'CENTER'),
('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
('FONTSIZE', (0, 0), (-1, 0), 12),
('BOTTOMPADDING', (0, 0), (-1, 0), 12),
('BACKGROUND', (0, 1), (-1, -2), colors.lightgrey),
('BACKGROUND', (0, -1), (-1, -1), colors.yellow),
('FONTNAME', (0, -1), (-1, -1), 'Helvetica-Bold'),
('GRID', (0, 0), (-1, -1), 1, colors.black),
]))
elements.append(expense_table)
elements.append(Spacer(1, 0.3*inch))
# Key Financial Metrics
elements.append(Paragraph("Key Financial Metrics", heading_style))
metrics_text = """
• Gross Profit Margin: 37.2%<br/>
• Operating Profit Margin: 18.4%<br/>
• Net Profit Margin: 14.8%<br/>
• EBITDA: $10,250,000 USD<br/>
• Cash Flow from Operations: $8,930,000 USD<br/>
• Total Assets: $125,400,000 USD<br/>
• Total Liabilities: $48,200,000 USD<br/>
• Shareholders' Equity: $77,200,000 USD<br/>
"""
elements.append(Paragraph(metrics_text, styles['BodyText']))
# Add page break
elements.append(PageBreak())
# Currency Exchange Rates Used
elements.append(Paragraph("Currency Exchange Rates (as of March 31, 2024)", heading_style))
exchange_data = [
['Currency Pair', 'Exchange Rate', 'Previous Quarter', 'Change'],
['USD/EUR', '0.9234', '0.9156', '+0.85%'],
['USD/GBP', '0.7891', '0.7823', '+0.87%'],
['USD/JPY', '149.85', '147.23', '+1.78%'],
['USD/SGD', '1.3452', '1.3389', '+0.47%'],
]
exchange_table = Table(exchange_data, colWidths=[2*inch, 1.5*inch, 1.5*inch, 1.2*inch])
exchange_table.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, 0), colors.grey),
('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke),
('ALIGN', (0, 0), (-1, -1), 'CENTER'),
('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
('FONTSIZE', (0, 0), (-1, 0), 12),
('BOTTOMPADDING', (0, 0), (-1, 0), 12),
('BACKGROUND', (0, 1), (-1, -1), colors.beige),
('GRID', (0, 0), (-1, -1), 1, colors.black),
]))
elements.append(exchange_table)
elements.append(Spacer(1, 0.3*inch))
# Investment Portfolio
elements.append(Paragraph("Investment Portfolio Performance", heading_style))
portfolio_text = """The company's investment portfolio showed strong performance in Q1 2024:
• Fixed Income Securities: $23,450,000 USD (yielding 4.2% annually)
• Equity Investments: $18,750,000 USD (up 8.3% this quarter)
• Real Estate Holdings: $31,200,000 USD (appreciation of 3.1%)
• Cash and Cash Equivalents: $15,890,000 USD
Total portfolio value: $89,290,000 USD, representing a 5.7% increase from Q4 2023."""
elements.append(Paragraph(portfolio_text, styles['BodyText']))
elements.append(Spacer(1, 0.3*inch))
# Future Projections
elements.append(Paragraph("Q2 2024 Projections", heading_style))
projection_data = [
['Metric', 'Q1 2024 Actual', 'Q2 2024 Projected', 'Growth'],
['Total Revenue', '$45,800,000', '$48,500,000', '+5.9%'],
['Operating Expenses', '$28,730,000', '$29,800,000', '+3.7%'],
['Net Income', '$6,780,000', '$7,450,000', '+9.9%'],
['EPS (Earnings Per Share)', '$2.34', '$2.57', '+9.8%'],
]
projection_table = Table(projection_data, colWidths=[2.5*inch, 1.5*inch, 1.5*inch, 1*inch])
projection_table.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, 0), colors.grey),
('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke),
('ALIGN', (0, 0), (-1, -1), 'CENTER'),
('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
('FONTSIZE', (0, 0), (-1, 0), 12),
('BOTTOMPADDING', (0, 0), (-1, 0), 12),
('BACKGROUND', (0, 1), (-1, -1), colors.lightblue),
('GRID', (0, 0), (-1, -1), 1, colors.black),
]))
elements.append(projection_table)
elements.append(Spacer(1, 0.3*inch))
# Footer
footer_text = """
<para alignment="center">
<b>Note:</b> All financial figures are preliminary and subject to audit.<br/>
For more information, please contact: investor.relations@globalcorp.com<br/>
Global Corporation © 2024 - Confidential Financial Report
</para>
"""
elements.append(Spacer(1, 0.5*inch))
elements.append(Paragraph(footer_text, styles['Normal']))
# Build PDF
doc.build(elements)
print(f"Sample PDF created: {filename}")
return filename
def create_simple_expense_report():
"""Create a simpler expense report for quick testing"""
filename = "simple_expense_report.pdf"
doc = SimpleDocTemplate(filename, pagesize=A4)
elements = []
styles = getSampleStyleSheet()
# Title
elements.append(Paragraph("Quarterly Expense Report", styles['Title']))
elements.append(Spacer(1, 0.2*inch))
# Simple expense data
elements.append(Paragraph("Q1 2024 Regional Expenses", styles['Heading2']))
expense_text = """
Our company has the following expenses for Q1 2024:
<b>United States Office:</b> $2,500,000 USD<br/>
<b>United Kingdom Office:</b> £1,800,000 GBP<br/>
<b>Japan Office:</b> ¥380,000,000 JPY<br/>
<b>European Union Office:</b> €2,100,000 EUR<br/>
<b>Singapore Office:</b> S$3,200,000 SGD<br/>
These expenses include salaries, operations, marketing, and R&D costs.
Additional financial metrics:
• Total headcount: 1,250 employees globally
• Average expense per employee: varies by region
• Projected Q2 expense reduction target: 8% across all regions
"""
elements.append(Paragraph(expense_text, styles['BodyText']))
# Build PDF
doc.build(elements)
print(f"Simple PDF created: {filename}")
return filename
if __name__ == "__main__":
# Create both PDFs
create_financial_report()
create_simple_expense_report()
# Create a test directory for PDFs if needed
os.makedirs("test_pdfs", exist_ok=True)
# Move PDFs to test directory
import shutil
for pdf in ["sample_financial_report_q1_2024.pdf", "simple_expense_report.pdf"]:
if os.path.exists(pdf):
shutil.move(pdf, f"test_pdfs/{pdf}")
print("\nPDFs created in test_pdfs/ directory")
print("You can host these PDFs online or use a local server for testing")
demo_conversation.py¶
#!/usr/bin/env python3
"""
Demo script showing conversation history persistence
"""
import os
from dotenv import load_dotenv
from agent import ContextAwareAgent, ContextMode
# Load environment variables
load_dotenv()
def main():
# Get API key (use any available provider)
if os.getenv("ARK_API_KEY"):
api_key, provider = os.getenv("ARK_API_KEY"), "doubao"
elif os.getenv("MOONSHOT_API_KEY"):
api_key, provider = os.getenv("MOONSHOT_API_KEY"), "kimi"
elif os.getenv("DEEPSEEK_API_KEY"):
api_key, provider = os.getenv("DEEPSEEK_API_KEY"), "deepseek"
elif os.getenv("SILICONFLOW_API_KEY"):
api_key, provider = os.getenv("SILICONFLOW_API_KEY"), "siliconflow"
else:
api_key, provider = None, None
if not api_key:
print("❌ No API key found. Please set one of:")
print(" - ARK_API_KEY")
print(" - MOONSHOT_API_KEY")
print(" - DEEPSEEK_API_KEY")
print(" - SILICONFLOW_API_KEY")
return
print("🎭 Conversation History Demo")
print("=" * 50)
print(f"Provider: {provider.upper()}")
print("-" * 50)
# Create agent
agent = ContextAwareAgent(
api_key=api_key,
provider=provider,
context_mode=ContextMode.FULL,
verbose=False
)
# Conversation 1: Set context
print("\n💬 Turn 1: Setting context...")
result = agent.execute_task("My name is Alice and I have a budget of $5,000. What is 20% of my budget?")
print(f"Agent: {result.get('final_answer', 'No answer')}")
# Conversation 2: Reference previous context
print("\n💬 Turn 2: Referencing previous context...")
result = agent.execute_task("Convert that 20% amount to EUR please.")
print(f"Agent: {result.get('final_answer', 'No answer')}")
# Conversation 3: Recall information
print("\n💬 Turn 3: Recalling information...")
result = agent.execute_task("What was my name and total budget that I mentioned?")
print(f"Agent: {result.get('final_answer', 'No answer')}")
print("\n" + "-" * 50)
print(f"📊 Final Statistics:")
print(f" Total messages in history: {len(agent.conversation_history)}")
print(f" Total tool calls made: {len(agent.trajectory.tool_calls)}")
# Show that system prompt is unchanged
system_prompt = agent.conversation_history[0]['content']
if "Alice" not in system_prompt and "5000" not in system_prompt:
print(" ✅ System prompt remained unchanged")
else:
print(" ❌ System prompt was modified")
print("\n✨ Demo complete!")
if __name__ == "__main__":
main()
demo_samples.py¶
#!/usr/bin/env python3
"""
Demo script to showcase sample tasks with PDF functionality
"""
import os
import sys
from pathlib import Path
from main import get_sample_tasks, ensure_sample_pdfs
def main():
"""Demo the sample tasks"""
print("\n" + "="*60)
print("🎯 CONTEXT-AWARE AGENT - SAMPLE TASKS DEMO")
print("="*60)
# Ensure PDFs exist
print("\n📄 Checking for sample PDFs...")
if ensure_sample_pdfs():
print("✅ Sample PDFs are ready!")
else:
print("⚠️ Could not create sample PDFs, will use online alternatives")
# Get sample tasks
tasks = get_sample_tasks()
print(f"\n📋 Found {len(tasks)} sample tasks:")
print("-"*60)
for i, task in enumerate(tasks, 1):
print(f"\n{i}. {task['name']}")
print(f" 📝 {task['description']}")
print(f" 📊 Complexity: {'⭐' * (i if i <= 3 else 3)}")
# Show a preview of the task
task_preview = task['task'].replace('\n', ' ')[:100] + "..."
print(f" 💬 Preview: {task_preview}")
print("\n" + "="*60)
print("💡 USAGE TIPS:")
print("-"*60)
print("1. Run 'python main.py' to enter interactive mode")
print("2. Type 'sample 3' to test PDF parsing capabilities")
print("3. Type 'sample 5' for the most comprehensive test")
print("4. Switch modes with 'mode no_reasoning' to see ablation effects")
print("\n" + "="*60)
print("🔬 ABLATION TESTING:")
print("-"*60)
print("Try running the same task in different modes:")
print(" • full - Everything works perfectly")
print(" • no_history - Agent forgets what it did")
print(" • no_reasoning - No planning, chaotic execution")
print(" • no_tool_calls - Can't do anything!")
print(" • no_tool_results - Works blind, gets confused")
print("\n" + "="*60)
print("📊 PDF TASKS:")
print("-"*60)
# Check if local PDFs exist
pdf_dir = Path("test_pdfs")
if pdf_dir.exists():
pdfs = list(pdf_dir.glob("*.pdf"))
if pdfs:
print(f"✅ Found {len(pdfs)} local PDF files:")
for pdf in pdfs:
print(f" • {pdf.name}")
print("\nTask #3 will use these local PDFs for testing.")
else:
print("⚠️ No PDFs found in test_pdfs/")
else:
print("📥 PDF directory not found. Run 'create_pdfs' command to generate samples.")
print("\n" + "="*60)
print("Ready to test! Run 'python main.py' to start.")
print("="*60 + "\n")
if __name__ == "__main__":
main()
main.py¶
"""
Main entry point for Context-Aware Agent
"""
import os
import sys
import argparse
import logging
from agent import ContextAwareAgent, ContextMode
import json
from pathlib import Path
import subprocess
import time
from typing import Dict, Any, List
from tabulate import tabulate
try:
import matplotlib.pyplot as plt
import numpy as np
MATPLOTLIB_AVAILABLE = True
except ImportError:
MATPLOTLIB_AVAILABLE = False
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Load .env so API keys configured there are available via os.getenv.
# (config.py calls load_dotenv() too, but main.py only imports agent, which
# does not import config, so we must trigger it here.)
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
pass
class AblationTestSuite:
"""Test suite for exploring context importance through ablation studies"""
def __init__(self, api_key: str, provider: str = "siliconflow", model: str = None):
"""
Initialize test suite
Args:
api_key: API key for the LLM provider
provider: LLM provider to use
model: Optional model override
"""
self.api_key = api_key
self.provider = provider
self.model = model
self.test_results = []
def create_complex_financial_task(self) -> str:
"""
Create a complex task that requires multiple tool calls and reasoning
Returns:
Task description
"""
return """Analyze the financial report from this PDF: https://www.berkshirehathaway.com/qtrly/1stqtr23.pdf
Please complete the following analysis:
1. Extract the total revenue figures from Q1 2023
2. Convert the revenue from USD to EUR, GBP, and JPY
3. Calculate the following metrics:
- Average revenue across the three converted currencies
- Percentage difference between highest and lowest converted values
- If the company maintains a 15% profit margin, what would be the profit in each currency?
Provide a comprehensive financial summary with all calculations shown."""
def create_multinational_budget_task(self) -> str:
"""
Create a task requiring multiple currency conversions and calculations
Returns:
Task description
"""
return """A multinational company has the following Q1 2024 expenses documented in this report:
https://github.com/adobe/pdf-services-node-sdk-samples/raw/refs/heads/master/resources/extractPDFInput.pdf
Tasks to complete:
1. Parse the PDF and extract all monetary values mentioned
2. The company operates in 5 regions with expenses in different currencies:
- US Office: $2,500,000 USD
- UK Office: £1,800,000 GBP
- Japan Office: ¥380,000,000 JPY
- EU Office: €2,100,000 EUR
- Singapore Office: $3,200,000 SGD
3. Convert all expenses to USD for consolidation
4. Calculate:
- Total global expenses in USD
- Average expense per region
- What percentage each region represents of total expenses
- If we apply a 8% cost reduction uniformly, what would be the new expense for each region in their local currency?
Present a detailed financial analysis with all conversions and calculations."""
def run_single_test(self, task: str, context_mode: ContextMode, test_name: str,
case_name: str = "default") -> Dict[str, Any]:
"""
Run a single ablation test
Args:
task: Task to execute
context_mode: Context mode to test
test_name: Name of the test
case_name: Name of the case/task this test belongs to
Returns:
Test results
"""
logger.info(f"\n{'='*60}")
logger.info(f"Running test: {test_name} [case: {case_name}]")
logger.info(f"Context mode: {context_mode.value}")
logger.info(f"{'='*60}")
agent = ContextAwareAgent(self.api_key, context_mode, provider=self.provider, model=self.model)
start_time = time.time()
result = agent.execute_task(task)
execution_time = time.time() - start_time
# Analyze the result
test_result = {
"test_name": test_name,
"case_name": case_name,
"context_mode": context_mode.value,
"execution_time": round(execution_time, 2),
"iterations": result.get("iterations", 0),
"num_tool_calls": len(result["trajectory"].tool_calls),
"success": result.get("success", False),
"has_final_answer": result.get("final_answer") is not None,
"error": result.get("error"),
"reasoning_steps": len(result["trajectory"].reasoning_steps),
"final_answer_preview": (result.get("final_answer", "")[:200] + "...") if result.get("final_answer") else None
}
# Log summary
logger.info(f"Test completed in {test_result['execution_time']}s")
logger.info(f"Success: {test_result['success']}")
logger.info(f"Tool calls made: {test_result['num_tool_calls']}")
logger.info(f"Iterations: {test_result['iterations']}")
if test_result["error"]:
logger.error(f"Error occurred: {test_result['error']}")
return test_result
def run_ablation_study(self, context_modes: List[ContextMode] = None,
cases: List[Dict[str, str]] = None) -> List[Dict[str, Any]]:
"""
Run ablation study across specified context modes and one or more cases.
Args:
context_modes: List of context modes to test (defaults to all modes)
cases: List of {"name", "task"} dicts to run each mode against.
Defaults to a single multinational-budget case, which
preserves the original single-task behaviour.
Returns:
Flat list of test results (one entry per case x mode).
"""
# Default: single multinational-budget case (preserves prior behaviour).
if cases is None:
cases = [{
"name": "Multinational Budget",
"task": self.create_multinational_budget_task()
}]
# Map context modes to test names
mode_names = {
ContextMode.FULL: "Baseline - Full Context",
ContextMode.NO_HISTORY: "Ablation 1 - No Historical Tool Calls",
ContextMode.NO_REASONING: "Ablation 2 - No Reasoning Process",
ContextMode.NO_TOOL_CALLS: "Ablation 3 - No Tool Call Commands",
ContextMode.NO_TOOL_RESULTS: "Ablation 4 - No Tool Call Results"
}
if context_modes is None:
context_modes = list(mode_names.keys())
results = []
for case in cases:
case_name = case["name"]
task = case["task"]
for context_mode in context_modes:
test_name = mode_names[context_mode]
try:
result = self.run_single_test(task, context_mode, test_name, case_name=case_name)
results.append(result)
self.test_results.append(result)
# Add delay between tests to avoid rate limiting
time.sleep(2)
except Exception as e:
logger.error(f"Failed to run test {test_name} [case: {case_name}]: {str(e)}")
results.append({
"test_name": test_name,
"case_name": case_name,
"context_mode": context_mode.value,
"error": str(e),
"success": False
})
return results
def analyze_results(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Analyze ablation study results
Args:
results: List of test results
Returns:
Analysis summary
"""
analysis = {
"total_tests": len(results),
"successful_tests": sum(1 for r in results if r.get("success", False)),
"context_mode_impact": {}
}
# Analyze impact of each ablation
baseline = next((r for r in results if r["context_mode"] == "full"), None)
if baseline:
for result in results:
if result["context_mode"] != "full":
mode_analysis = {
"success_maintained": result.get("success", False),
"execution_time_delta": result.get("execution_time", 0) - baseline.get("execution_time", 0),
"iteration_delta": result.get("iterations", 0) - baseline.get("iterations", 0),
"tool_call_delta": result.get("num_tool_calls", 0) - baseline.get("num_tool_calls", 0),
"failure_reason": None
}
# Identify failure reasons
if not result.get("success", False):
if result["context_mode"] == "no_tool_calls":
mode_analysis["failure_reason"] = "Cannot execute tools without tool call capability"
elif result["context_mode"] == "no_tool_results":
mode_analysis["failure_reason"] = "Cannot make informed decisions without tool results"
elif result["context_mode"] == "no_history":
mode_analysis["failure_reason"] = "May repeat actions or lose track of progress"
elif result["context_mode"] == "no_reasoning":
mode_analysis["failure_reason"] = "Lacks planning and strategic thinking"
analysis["context_mode_impact"][result["context_mode"]] = mode_analysis
return analysis
def print_results_table(self, results: List[Dict[str, Any]]):
"""
Print results in a formatted table
Args:
results: List of test results
"""
# Prepare data for tabulation
table_data = []
for result in results:
table_data.append([
result.get("case_name", "default"),
result["context_mode"],
"✓" if result.get("success", False) else "✗",
f"{result.get('execution_time', 0)}s",
result.get("iterations", 0),
result.get("num_tool_calls", 0),
result.get("reasoning_steps", 0),
"Yes" if result.get("has_final_answer", False) else "No"
])
headers = ["Case", "Context Mode", "Success", "Time", "Iterations", "Tool Calls", "Reasoning Steps", "Final Answer"]
print("\n" + "="*80)
print("ABLATION STUDY RESULTS")
print("="*80)
print(tabulate(table_data, headers=headers, tablefmt="grid"))
def print_comparison_matrix(self, results: List[Dict[str, Any]]):
"""
Print a mode x case comparison matrix so the effect of each context
component can be read at a glance across every case.
Args:
results: List of test results
"""
cases = []
for r in results:
c = r.get("case_name", "default")
if c not in cases:
cases.append(c)
modes = []
for r in results:
m = r["context_mode"]
if m not in modes:
modes.append(m)
# Index results by (mode, case) for quick lookup.
by_key = {(r["context_mode"], r.get("case_name", "default")): r for r in results}
table_data = []
for mode in modes:
row = [mode]
for case in cases:
r = by_key.get((mode, case))
if r is None:
row.append("-")
elif r.get("success", False):
row.append(f"✓ {r.get('iterations', 0)}it/{r.get('num_tool_calls', 0)}tc")
else:
row.append(f"✗ {r.get('iterations', 0)}it/{r.get('num_tool_calls', 0)}tc")
table_data.append(row)
headers = ["Context Mode"] + cases
print("\n" + "="*80)
print("COMPARISON MATRIX (rows = context mode, cols = case; cell = success it=iterations tc=tool calls)")
print("="*80)
print(tabulate(table_data, headers=headers, tablefmt="grid"))
def visualize_results(self, results: List[Dict[str, Any]]):
"""
Create visualizations of ablation study results
Args:
results: List of test results
"""
if not MATPLOTLIB_AVAILABLE:
logger.warning("Matplotlib not available, skipping visualizations")
return
# Extract data for visualization
modes = [r["context_mode"] for r in results]
iterations = [r.get("iterations", 0) for r in results]
tool_calls = [r.get("num_tool_calls", 0) for r in results]
exec_times = [r.get("execution_time", 0) for r in results]
success = [1 if r.get("success", False) else 0 for r in results]
# Create subplots
fig, axes = plt.subplots(2, 2, figsize=(12, 10))
fig.suptitle("Ablation Study: Impact of Context Components", fontsize=16)
# Plot 1: Success Rate
axes[0, 0].bar(modes, success, color=['green' if s else 'red' for s in success])
axes[0, 0].set_title("Task Success by Context Mode")
axes[0, 0].set_ylabel("Success (1) / Failure (0)")
axes[0, 0].tick_params(axis='x', rotation=45)
# Plot 2: Iterations Required
axes[0, 1].bar(modes, iterations, color='blue')
axes[0, 1].set_title("Iterations Required")
axes[0, 1].set_ylabel("Number of Iterations")
axes[0, 1].tick_params(axis='x', rotation=45)
# Plot 3: Tool Calls Made
axes[1, 0].bar(modes, tool_calls, color='orange')
axes[1, 0].set_title("Tool Calls Made")
axes[1, 0].set_ylabel("Number of Tool Calls")
axes[1, 0].tick_params(axis='x', rotation=45)
# Plot 4: Execution Time
axes[1, 1].bar(modes, exec_times, color='purple')
axes[1, 1].set_title("Execution Time")
axes[1, 1].set_ylabel("Time (seconds)")
axes[1, 1].tick_params(axis='x', rotation=45)
plt.tight_layout()
plt.savefig("ablation_study_results.png", dpi=150, bbox_inches='tight')
logger.info("Visualization saved as 'ablation_study_results.png'")
def generate_report(self, results: List[Dict[str, Any]], analysis: Dict[str, Any]) -> str:
"""
Generate a comprehensive report of the ablation study
Args:
results: List of test results
analysis: Analysis summary
Returns:
Report text
"""
report = """
# Context Ablation Study Report
## Executive Summary
This ablation study explores the critical importance of different context components in AI agent behavior.
## Test Configuration
- **Provider**: {provider}
- **Model**: {model}
- **Task**: Complex financial analysis requiring PDF parsing, currency conversion, and calculations
- **Context Modes Tested**: {num_modes}
## Key Findings
### 1. Complete Lack of Historical Tool Calls (NO_HISTORY)
**Impact**: Agent loses track of previous actions and may repeat operations unnecessarily.
- **Behavior**: Agent cannot reference past tool executions, leading to redundant API calls
- **Performance**: {no_history_perf}
- **Critical for**: Multi-step tasks requiring sequential dependencies
### 2. Lack of Reasoning Process (NO_REASONING)
**Impact**: Agent operates without strategic planning or step-by-step thinking.
- **Behavior**: Direct execution without planning leads to inefficient or incorrect solutions
- **Performance**: {no_reasoning_perf}
- **Critical for**: Complex tasks requiring logical decomposition
### 3. Lack of Tool Call Commands (NO_TOOL_CALLS)
**Impact**: Agent cannot execute any external tools, rendering it unable to complete the task.
- **Behavior**: Complete failure - agent can only describe what it would do, not actually do it
- **Performance**: {no_tool_calls_perf}
- **Critical for**: Any task requiring external data or computation
### 4. Lack of Tool Call Results (NO_TOOL_RESULTS)
**Impact**: Agent operates blind to the outcomes of its actions.
- **Behavior**: Cannot adapt based on results, leading to incorrect conclusions
- **Performance**: {no_tool_results_perf}
- **Critical for**: Tasks requiring iterative refinement or result validation
## Statistical Summary
- **Total Tests Run**: {total_tests}
- **Successful Completions**: {successful_tests}
- **Average Execution Time (Full Context)**: {avg_exec_time}s
- **Average Tool Calls (Full Context)**: {avg_tool_calls}
## Conclusion
The ablation study clearly demonstrates that each context component plays a vital role:
1. **Tool calls** are fundamental - without them, the agent cannot interact with the world
2. **Tool results** provide essential feedback for decision-making
3. **Reasoning** enables strategic planning and efficient execution
4. **History** prevents redundancy and maintains task coherence
## Recommendations
- Always maintain complete context for production agents
- Consider context windowing rather than removal for memory optimization
- Implement fallback mechanisms when context components are unavailable
"""
# Fill in performance metrics
def get_perf_string(mode):
mode_result = next((r for r in results if r["context_mode"] == mode), None)
if mode_result:
return f"{'SUCCESS' if mode_result.get('success', False) else 'FAILURE'} - {mode_result.get('iterations', 0)} iterations, {mode_result.get('execution_time', 0)}s"
return "N/A"
# Calculate baseline metrics
baseline = next((r for r in results if r["context_mode"] == "full"), None)
avg_exec_time = baseline.get("execution_time", 0) if baseline else 0
avg_tool_calls = baseline.get("num_tool_calls", 0) if baseline else 0
report = report.format(
provider=self.provider,
model=self.model or "default",
num_modes=len(results),
no_history_perf=get_perf_string("no_history"),
no_reasoning_perf=get_perf_string("no_reasoning"),
no_tool_calls_perf=get_perf_string("no_tool_calls"),
no_tool_results_perf=get_perf_string("no_tool_results"),
total_tests=analysis["total_tests"],
successful_tests=analysis["successful_tests"],
avg_exec_time=avg_exec_time,
avg_tool_calls=avg_tool_calls
)
return report
def run_single_task(api_key: str, task: str, context_mode: str = "full", provider: str = "siliconflow", model: str = None, output: str = None):
"""
Run a single task with the agent
Args:
api_key: API key for the LLM provider
task: Task description
context_mode: Context mode to use
provider: LLM provider to use
model: Optional model override
output: Optional path for the JSON result (default task_result_{mode}.json)
"""
# Parse context mode
mode_map = {
"full": ContextMode.FULL,
"no_history": ContextMode.NO_HISTORY,
"no_reasoning": ContextMode.NO_REASONING,
"no_tool_calls": ContextMode.NO_TOOL_CALLS,
"no_tool_results": ContextMode.NO_TOOL_RESULTS
}
if context_mode not in mode_map:
logger.error(f"Invalid context mode: {context_mode}")
logger.info(f"Valid modes: {', '.join(mode_map.keys())}")
return
# Create agent
agent = ContextAwareAgent(api_key, mode_map[context_mode], provider=provider, model=model)
logger.info(f"Running task with context mode: {context_mode}")
logger.info(f"Task: {task[:100]}...")
# Execute task
result = agent.execute_task(task)
# Print results
print("\n" + "="*60)
print("TASK EXECUTION RESULT")
print("="*60)
print(f"Context Mode: {context_mode}")
print(f"Success: {result.get('success', False)}")
print(f"Iterations: {result.get('iterations', 0)}")
print(f"Tool Calls: {len(result['trajectory'].tool_calls)}")
if result.get('final_answer'):
print(f"\nFinal Answer:")
print("-"*40)
print(result['final_answer'])
if result.get('error'):
print(f"\nError: {result['error']}")
# Save detailed results
output_file = output or f"task_result_{context_mode}.json"
with open(output_file, 'w') as f:
# Convert trajectory to serializable format
serializable_result = {
"success": result.get("success", False),
"iterations": result.get("iterations", 0),
"final_answer": result.get("final_answer"),
"error": result.get("error"),
"context_mode": context_mode,
"tool_calls": [
{
"tool_name": tc.tool_name,
"arguments": tc.arguments,
"result": tc.result,
"timestamp": tc.timestamp
}
for tc in result["trajectory"].tool_calls
],
"reasoning_steps": result["trajectory"].reasoning_steps
}
json.dump(serializable_result, f, indent=2)
logger.info(f"Detailed results saved to {output_file}")
def ensure_sample_pdfs():
"""
Ensure sample PDFs exist, create them if they don't
Returns:
bool: True if PDFs are available
"""
pdf_dir = Path("test_pdfs")
sample_pdf = pdf_dir / "simple_expense_report.pdf"
if not pdf_dir.exists() or not sample_pdf.exists():
print("\n📚 Sample PDFs not found. Creating them...")
try:
# Run the PDF creation script
result = subprocess.run(
[sys.executable, "create_sample_pdf.py"],
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0:
print("✅ Sample PDFs created successfully in test_pdfs/")
return True
else:
print(f"⚠️ Could not create PDFs: {result.stderr}")
return False
except Exception as e:
print(f"⚠️ Error creating PDFs: {str(e)}")
return False
return True
def get_sample_tasks():
"""
Get sample tasks for testing
Returns:
list: List of sample task dictionaries
"""
# Check if we're running locally or need to use online PDFs
local_pdfs = Path("test_pdfs").exists()
if local_pdfs:
pdf_path = "file://" + str(Path.cwd() / "test_pdfs" / "simple_expense_report.pdf")
pdf_note = "Using local PDF"
else:
# Use a publicly available PDF for testing
pdf_path = "https://www.w3.org/WAI/ER/tests/xhtml/testfiles/resources/pdf/dummy.pdf"
pdf_note = "Using online sample PDF"
return [
{
"name": "📊 Currency Conversion Task",
"description": "Convert between multiple currencies",
"task": """Convert $1000 USD to EUR, GBP, and JPY.
Then calculate the average value across all three converted currencies."""
},
{
"name": "📄 PDF Analysis Task",
"description": f"Extract and analyze data from PDF ({pdf_note})",
"task": f"""Analyze this PDF document: {pdf_path}
Extract any text content and provide a summary of what you found."""
},
{
"name": "💰 Complex Financial Analysis",
"description": "Multi-step financial calculation",
"task": """A company has the following quarterly revenues:
- Q1: $2,500,000 USD
- Q2: €2,100,000 EUR
- Q3: £1,800,000 GBP
- Q4: ¥380,000,000 JPY
Please:
1. Convert all revenues to USD
2. Calculate the total annual revenue in USD
3. Determine the average quarterly revenue
4. Find which quarter had the highest revenue
5. If the company has a 20% profit margin, calculate the annual profit in USD"""
},
{
"name": "🌍 Multi-Currency Budget Planning",
"description": "International budget calculations",
"task": """An international conference has the following budget allocations:
- Venue (UK): £45,000
- Speakers (US): $75,000
- Catering (France): €38,000
- Technology (Japan): ¥8,500,000
- Marketing (Singapore): S$25,000
Tasks:
1. Convert all amounts to USD
2. Calculate the total budget
3. Determine what percentage each category represents
4. If we need to cut the budget by 15%, how much should each category be reduced to (in their original currencies)?"""
},
{
"name": "📈 Investment Portfolio Analysis",
"description": "Analyze international investment returns",
"task": """An investor has the following international investments with their current values:
- US Tech Stocks: $125,000 (purchased for $100,000)
- European Bonds: €85,000 (purchased for €90,000)
- UK Real Estate: £200,000 (purchased for £175,000)
- Japanese ETFs: ¥15,000,000 (purchased for ¥12,000,000)
Calculate:
1. Convert all current values to USD
2. Convert all purchase prices to USD (use current exchange rates for simplicity)
3. Calculate the profit/loss for each investment in USD
4. Determine the total portfolio value and overall return percentage
5. Which investment performed best in percentage terms?"""
}
]
def get_ablation_cases(num_cases: int = 1) -> List[Dict[str, str]]:
"""
Build a list of self-contained cases for the ablation study.
Reuses the pre-defined sample tasks, skipping the PDF task (which needs
network access) so the study is reproducible offline apart from the LLM
call itself. num_cases=1 returns a single case; larger values add more.
Args:
num_cases: How many cases to include (clamped to [1, available]).
Returns:
List of {"name", "task"} dicts.
"""
samples = get_sample_tasks()
# Indices 0/2/3/4 are currency/finance tasks (index 1 is the PDF task).
candidates = [samples[0], samples[2], samples[3], samples[4]]
num_cases = max(1, min(num_cases, len(candidates)))
return [{"name": c["name"], "task": c["task"]} for c in candidates[:num_cases]]
def run_ablation_study(api_key: str, provider: str = "siliconflow", model: str = None,
context_modes: List[str] = None, num_cases: int = 1,
output: str = None):
"""
Run ablation study to test importance of context
Args:
api_key: API key for the LLM provider
provider: LLM provider to use
model: Optional model override
context_modes: List of context mode names to test (defaults to all)
num_cases: Number of cases to run each mode against. 1 (default) keeps
the original single multinational-budget task; >1 runs a multi-case
comparison across the sample tasks.
output: Optional path for the raw JSON results (default ablation_results.json)
"""
# Parse context modes if provided
mode_map = {
"full": ContextMode.FULL,
"no_history": ContextMode.NO_HISTORY,
"no_reasoning": ContextMode.NO_REASONING,
"no_tool_calls": ContextMode.NO_TOOL_CALLS,
"no_tool_results": ContextMode.NO_TOOL_RESULTS
}
modes_to_test = None
if context_modes:
modes_to_test = []
for mode_name in context_modes:
if mode_name in mode_map:
modes_to_test.append(mode_map[mode_name])
else:
logger.error(f"Invalid context mode: {mode_name}")
logger.info(f"Valid modes: {', '.join(mode_map.keys())}")
return
# Build cases. num_cases <= 1 keeps the original single-case behaviour.
cases = None
if num_cases and num_cases > 1:
cases = get_ablation_cases(num_cases)
test_suite = AblationTestSuite(api_key, provider=provider, model=model)
logger.info("Starting ablation study...")
if modes_to_test:
logger.info(f"Testing modes: {', '.join(context_modes)}")
else:
logger.info("Testing all context modes")
logger.info(f"Cases: {len(cases) if cases else 1}")
results = test_suite.run_ablation_study(modes_to_test, cases=cases)
# Analyze results
analysis = test_suite.analyze_results(results)
# Print results table
test_suite.print_results_table(results)
# Print mode x case comparison matrix
test_suite.print_comparison_matrix(results)
# Generate visualizations
try:
test_suite.visualize_results(results)
except Exception as e:
logger.warning(f"Could not generate visualizations: {str(e)}")
# Generate and save report
report = test_suite.generate_report(results, analysis)
with open("ablation_study_report.md", "w") as f:
f.write(report)
logger.info("Report saved as 'ablation_study_report.md'")
# Save raw results
output_file = output or "ablation_results.json"
with open(output_file, "w") as f:
json.dump(results, f, indent=2)
logger.info(f"Raw results saved as '{output_file}'")
# Print analysis summary
print("\n" + "="*80)
print("ANALYSIS SUMMARY")
print("="*80)
print(f"Success Rate: {analysis['successful_tests']}/{analysis['total_tests']}")
print("\nContext Mode Impacts:")
for mode, impact in analysis["context_mode_impact"].items():
print(f"\n{mode.upper()}:")
print(f" - Success Maintained: {impact['success_maintained']}")
print(f" - Execution Time Delta: {impact['execution_time_delta']:.2f}s")
print(f" - Iteration Delta: {impact['iteration_delta']}")
print(f" - Tool Call Delta: {impact['tool_call_delta']}")
if impact['failure_reason']:
print(f" - Failure Reason: {impact['failure_reason']}")
def interactive_mode(api_key: str, provider: str = "siliconflow", model: str = None):
"""
Run the agent in interactive mode
Args:
api_key: API key for the LLM provider
provider: LLM provider to use
model: Optional model override
"""
# Store current provider and model
current_provider = provider
current_model = model
current_api_key = api_key
# Available providers
available_providers = ["siliconflow", "doubao", "kimi", "moonshot", "deepseek"]
print("\n" + "="*60)
print("INTERACTIVE MODE - Context-Aware Agent")
print(f"Provider: {current_provider.upper()} | Model: {current_model or 'default'}")
print("="*60)
print("Available commands:")
print(" - Type your task/question")
print(" - 'samples' to see sample tasks")
print(" - 'sample <number>' to run a sample task")
print(" - 'create_pdfs' to create sample PDF files")
print(" - 'providers' to list available providers")
print(" - 'provider <name>' to switch provider")
print(" - 'modes' to see available context modes")
print(" - 'mode <mode_name>' to switch context mode")
print(" - 'reset' to reset agent trajectory")
print(" - 'status' to show current configuration")
print(" - 'help' to show all commands")
print(" - 'quit' to exit")
print("-"*60)
# Ensure sample PDFs exist
ensure_sample_pdfs()
# Get sample tasks
sample_tasks = get_sample_tasks()
# Initialize agent with full context
mode_map = {
"full": ContextMode.FULL,
"no_history": ContextMode.NO_HISTORY,
"no_reasoning": ContextMode.NO_REASONING,
"no_tool_calls": ContextMode.NO_TOOL_CALLS,
"no_tool_results": ContextMode.NO_TOOL_RESULTS
}
current_mode = ContextMode.FULL
agent = ContextAwareAgent(current_api_key, current_mode, provider=current_provider, model=current_model)
while True:
try:
# Show current provider in prompt
prompt = f"\n[{current_provider.upper()}]> "
user_input = input(prompt).strip()
if user_input.lower() == 'quit':
print("Goodbye!")
break
elif user_input.lower() == 'help':
print("\n📚 Available Commands:")
print(" samples - Display all available sample tasks")
print(" sample <n> - Run sample task number n")
print(" providers - List all available LLM providers")
print(" provider <name> - Switch to a different provider")
print(" modes - List available context modes")
print(" mode <name> - Switch context mode")
print(" status - Show current configuration")
print(" reset - Reset agent trajectory")
print(" create_pdfs - Generate sample PDF files")
print(" help - Show this help message")
print(" quit - Exit interactive mode")
print("\nOr type any task/question to execute it.")
elif user_input.lower() == 'samples':
print("\n📋 Available sample tasks:")
for i, sample in enumerate(sample_tasks, 1):
print(f"\n{i}. {sample['name']}")
print(f" {sample['description']}")
elif user_input.lower().startswith('sample '):
try:
sample_num = int(user_input.split()[1])
if 1 <= sample_num <= len(sample_tasks):
sample = sample_tasks[sample_num - 1]
print(f"\n📌 Running: {sample['name']}")
print(f"Task: {sample['task']}")
print("\nProcessing...")
result = agent.execute_task(sample['task'])
print(f"\n{'='*40}")
print(f"Success: {result.get('success', False)}")
print(f"Iterations: {result.get('iterations', 0)}")
print(f"Tool Calls: {len(result['trajectory'].tool_calls)}")
if result.get('final_answer'):
print(f"\nAnswer:")
print(result['final_answer'])
if result.get('error'):
print(f"\nError: {result['error']}")
else:
print(f"Invalid sample number. Use 'sample 1' through 'sample {len(sample_tasks)}'")
except (ValueError, IndexError):
print(f"Invalid sample number. Use 'sample 1' through 'sample {len(sample_tasks)}'")
elif user_input.lower() == 'create_pdfs':
print("\n📄 Creating sample PDFs...")
try:
result = subprocess.run(
[sys.executable, "create_sample_pdf.py"],
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0:
print("✅ Sample PDFs created successfully in test_pdfs/")
# Update sample tasks with new PDF paths
sample_tasks = get_sample_tasks()
else:
print(f"⚠️ Could not create PDFs: {result.stderr}")
except Exception as e:
print(f"⚠️ Error creating PDFs: {str(e)}")
elif user_input.lower() == 'providers':
print("\n🔌 Available providers:")
for p in available_providers:
status = " (current)" if p == current_provider else ""
if p == "siliconflow":
print(f" - siliconflow: Qwen model{status}")
elif p == "doubao":
print(f" - doubao: ByteDance model{status}")
elif p in ["kimi", "moonshot"]:
print(f" - {p}: Moonshot Kimi K3 model{status}")
elif p == "deepseek":
print(f" - deepseek: DeepSeek V4 model{status}")
elif user_input.lower().startswith('provider '):
new_provider = user_input[9:].strip().lower()
if new_provider in available_providers:
# Get the appropriate API key for the new provider
if new_provider == "siliconflow":
new_api_key = os.getenv("SILICONFLOW_API_KEY")
if not new_api_key:
print("❌ SILICONFLOW_API_KEY not set in environment")
continue
elif new_provider == "doubao":
new_api_key = os.getenv("ARK_API_KEY")
if not new_api_key:
print("❌ ARK_API_KEY not set in environment")
continue
elif new_provider in ["kimi", "moonshot"]:
new_api_key = os.getenv("MOONSHOT_API_KEY")
if not new_api_key:
print("❌ MOONSHOT_API_KEY not set in environment")
continue
elif new_provider == "deepseek":
new_api_key = os.getenv("DEEPSEEK_API_KEY")
if not new_api_key:
print("❌ DEEPSEEK_API_KEY not set in environment")
continue
# Update current settings
current_provider = new_provider
current_api_key = new_api_key
current_model = None # Reset to use default model for new provider
# Create new agent with new provider
agent = ContextAwareAgent(current_api_key, current_mode, provider=current_provider, model=current_model)
# Get default model name from config
from config import Config
default_model = Config.get_default_model(current_provider)
print(f"✅ Switched to provider: {current_provider}")
print(f" Using model: {default_model}")
else:
print(f"❌ Invalid provider. Available: {', '.join(available_providers)}")
elif user_input.lower() == 'modes':
print("\n🔧 Available context modes:")
for mode in mode_map.keys():
print(f" - {mode}")
elif user_input.lower().startswith('mode '):
new_mode = user_input[5:].strip()
if new_mode in mode_map:
current_mode = mode_map[new_mode]
agent = ContextAwareAgent(current_api_key, current_mode, provider=current_provider, model=current_model)
print(f"✅ Switched to context mode: {current_mode.value}")
if current_mode != ContextMode.FULL:
print(f"⚠️ Warning: This mode intentionally disables certain features for testing")
else:
print(f"❌ Invalid mode. Available: {', '.join(mode_map.keys())}")
elif user_input.lower() == 'reset':
agent.reset()
print("✅ Agent trajectory and conversation history reset.")
elif user_input.lower() == 'status':
from config import Config
model_name = current_model or Config.get_default_model(current_provider)
print("\n📊 Current Configuration:")
print(f" Provider: {current_provider.upper()}")
print(f" Model: {model_name}")
print(f" Context Mode: {current_mode.value}")
print(f" Conversation History: {len(agent.conversation_history)} messages")
print(f" Tool Calls: {len(agent.trajectory.tool_calls)}")
# Show API key status
if current_provider == "siliconflow":
key_status = "✅ Set" if os.getenv("SILICONFLOW_API_KEY") else "❌ Not set"
print(f" API Key (SILICONFLOW_API_KEY): {key_status}")
elif current_provider == "doubao":
key_status = "✅ Set" if os.getenv("ARK_API_KEY") else "❌ Not set"
print(f" API Key (ARK_API_KEY): {key_status}")
elif current_provider in ["kimi", "moonshot"]:
key_status = "✅ Set" if os.getenv("MOONSHOT_API_KEY") else "❌ Not set"
print(f" API Key (MOONSHOT_API_KEY): {key_status}")
elif current_provider == "deepseek":
key_status = "✅ Set" if os.getenv("DEEPSEEK_API_KEY") else "❌ Not set"
print(f" API Key (DEEPSEEK_API_KEY): {key_status}")
elif user_input:
# Execute task
print("\nProcessing...")
result = agent.execute_task(user_input)
print(f"\n{'='*40}")
print(f"Success: {result.get('success', False)}")
print(f"Iterations: {result.get('iterations', 0)}")
print(f"Tool Calls: {len(result['trajectory'].tool_calls)}")
if result.get('final_answer'):
print(f"\nAnswer:")
print(result['final_answer'])
if result.get('error'):
print(f"\nError: {result['error']}")
except KeyboardInterrupt:
print("\n\nInterrupted. Type 'quit' to exit.")
except Exception as e:
print(f"Error: {str(e)}")
def main():
"""Main function"""
parser = argparse.ArgumentParser(
description="上下文感知 AI Agent（第一章 实验 1.1：上下文消融实验）",
epilog="""示例：
# 交互模式（默认）
python main.py
# 运行完整消融实验（五种上下文模式对照）
python main.py --mode ablation
# 多案例消融对照，输出结果到指定文件
python main.py --mode ablation --cases 3 --output my_ablation.json
# 只对比“完整上下文”与“无历史消息”两种模式
python main.py --mode ablation --ablation-modes full no_history
# 单任务执行，指定上下文模式与提供商
python main.py --mode single --task "把 1000 美元换算成欧元、英镑、日元并求平均值" \\
--context-mode full --provider doubao
""",
formatter_class=argparse.RawDescriptionHelpFormatter
)
parser.add_argument(
"--mode",
choices=["single", "ablation", "interactive"],
default="interactive",
help="运行模式：single=单任务，ablation=消融实验，interactive=交互式（默认）"
)
parser.add_argument(
"--task",
type=str,
help="要执行的任务/问题（用于 single 模式；不提供则从示例任务中选择）"
)
parser.add_argument(
"--context-mode",
choices=["full", "no_history", "no_reasoning", "no_tool_calls", "no_tool_results"],
default="full",
help="single 模式下的上下文模式：full=完整；no_history=无历史消息；"
"no_reasoning=无思考过程；no_tool_calls=无工具定义；no_tool_results=无工具执行结果"
)
parser.add_argument(
"--ablation-modes",
nargs="+",
choices=["full", "no_history", "no_reasoning", "no_tool_calls", "no_tool_results"],
help="消融实验中要测试的上下文模式（默认测试全部五种）"
)
parser.add_argument(
"--cases",
type=int,
default=1,
help="消融实验运行的案例数量（默认 1；>1 时在多个示例任务上做跨模式对照）"
)
parser.add_argument(
"--provider",
choices=["siliconflow", "doubao", "kimi", "moonshot", "deepseek", "openrouter"],
default="doubao",
help="LLM 提供商（默认：doubao；openrouter 或缺失主 key 时经 OpenRouter 兜底）"
)
parser.add_argument(
"--model",
type=str,
help="使用的模型名称（可选，不指定则使用该提供商的默认模型）"
)
parser.add_argument(
"--api-key",
type=str,
help="LLM 提供商的 API Key（也可通过环境变量 SILICONFLOW_API_KEY/ARK_API_KEY/MOONSHOT_API_KEY/DEEPSEEK_API_KEY 设置）"
)
parser.add_argument(
"--output",
type=str,
help="结果输出文件路径（single 模式为 JSON 结果，ablation 模式为原始结果 JSON）"
)
args = parser.parse_args()
# Get API key based on provider (with universal OpenRouter fallback)
if args.api_key:
api_key = args.api_key
elif args.provider == "openrouter":
api_key = os.getenv("OPENROUTER_API_KEY")
elif args.provider == "doubao":
api_key = os.getenv("ARK_API_KEY")
elif args.provider == "siliconflow":
api_key = os.getenv("SILICONFLOW_API_KEY")
elif args.provider in ["kimi", "moonshot"]:
api_key = os.getenv("MOONSHOT_API_KEY")
elif args.provider == "deepseek":
api_key = os.getenv("DEEPSEEK_API_KEY")
else:
logger.error(f"Unknown provider: {args.provider}")
sys.exit(1)
# If the primary provider key is missing, fall back to OpenRouter when set.
# An empty api_key lets ContextAwareAgent.resolve_llm_backend route via OpenRouter.
if not api_key:
if os.getenv("OPENROUTER_API_KEY"):
logger.info(
f"{args.provider} API key not set; falling back to OpenRouter "
"(OPENROUTER_API_KEY). Set the provider key to use it directly."
)
api_key = ""
else:
logger.error(
"No API key found. Set the provider key "
"(ARK_API_KEY/SILICONFLOW_API_KEY/MOONSHOT_API_KEY/DEEPSEEK_API_KEY) or "
"OPENROUTER_API_KEY (universal fallback)."
)
sys.exit(1)
# Log provider info
logger.info(f"Using provider: {args.provider}, model: {args.model or 'default'}")
# Execute based on mode
if args.mode == "single":
if not args.task:
# Prompt user to select a sample task
print("\n" + "="*60)
print("SINGLE TASK MODE - No task provided")
print("="*60)
# Ensure PDFs exist
ensure_sample_pdfs()
# Get and display sample tasks
sample_tasks = get_sample_tasks()
print("\n📋 Available sample tasks:")
for i, sample in enumerate(sample_tasks, 1):
print(f"\n{i}. {sample['name']}")
print(f" {sample['description']}")
print("\n" + "="*60)
try:
choice = input("\nSelect a task number (1-{}) or 'q' to quit: ".format(len(sample_tasks))).strip()
if choice.lower() == 'q':
sys.exit(0)
task_num = int(choice)
if 1 <= task_num <= len(sample_tasks):
selected_task = sample_tasks[task_num - 1]
print(f"\n✅ Selected: {selected_task['name']}")
print("\nTask details:")
print("-"*40)
print(selected_task['task'])
print("-"*40)
confirm = input("\nRun this task? (y/n): ").strip().lower()
if confirm == 'y':
run_single_task(api_key, selected_task['task'], args.context_mode,
provider=args.provider, model=args.model, output=args.output)
else:
print("Task cancelled.")
else:
print(f"Invalid selection. Please choose 1-{len(sample_tasks)}")
sys.exit(1)
except (ValueError, KeyboardInterrupt):
print("\nExiting...")
sys.exit(0)
else:
run_single_task(api_key, args.task, args.context_mode,
provider=args.provider, model=args.model, output=args.output)
elif args.mode == "ablation":
run_ablation_study(api_key, provider=args.provider, model=args.model,
context_modes=args.ablation_modes, num_cases=args.cases,
output=args.output)
else: # interactive
interactive_mode(api_key, provider=args.provider, model=args.model)
if __name__ == "__main__":
main()
quick_test_deepseek.py¶
#!/usr/bin/env python3
"""
Quick smoke test for DeepSeek provider (deepseek-v4-flash).
"""
import os
import sys
import time
from dotenv import load_dotenv
load_dotenv()
task = "What is 10 + 5? Provide FINAL ANSWER with just the number."
print("=" * 60)
print("QUICK TEST - DeepSeek Provider")
print("=" * 60)
deepseek_key = os.getenv("DEEPSEEK_API_KEY")
if not deepseek_key:
print("❌ DEEPSEEK_API_KEY not set")
print("Set it in .env or: export DEEPSEEK_API_KEY=your_key")
print("Get a key at: https://platform.deepseek.com/api_keys")
sys.exit(1)
from agent import ContextAwareAgent, ContextMode
agent = ContextAwareAgent(deepseek_key, ContextMode.FULL, provider="deepseek")
print(f"✅ Using: {agent.provider} / {agent.model}")
print(f" Base URL: {agent.client.base_url}")
print(f"\n📝 Task: {task}")
print("-" * 40)
start = time.time()
print("Processing...")
try:
result = agent.execute_task(task, max_iterations=3)
elapsed = time.time() - start
print(f"\n✅ Completed in {elapsed:.2f} seconds")
if result.get("success"):
print("Success: True")
if result.get("final_answer"):
print(f"Answer: {result['final_answer']}")
else:
print("Success: False")
if result.get("error"):
print(f"Error: {result['error']}")
print(f"Iterations: {result.get('iterations', 0)}")
print(f"Tool calls: {len(result['trajectory'].tool_calls)}")
except KeyboardInterrupt:
print("\n⚠️ Interrupted")
except Exception as e:
print(f"\n❌ Error: {str(e)}")
sys.exit(1)
print("=" * 60)
quick_test_doubao.py¶
#!/usr/bin/env python3
"""
Quick test with Doubao as default
"""
import os
import sys
import time
# Set a very simple task to test quickly
task = "What is 10 + 5? Provide FINAL ANSWER with just the number."
print("="*60)
print("QUICK TEST - Doubao Default Provider")
print("="*60)
ark_key = os.getenv("ARK_API_KEY")
if not ark_key:
print("❌ ARK_API_KEY not set")
sys.exit(1)
from agent import ContextAwareAgent, ContextMode
# Create agent with default Doubao
agent = ContextAwareAgent(ark_key, ContextMode.FULL, provider="doubao")
print(f"✅ Using: {agent.provider} / {agent.model}")
print(f"\n📝 Task: {task}")
print("-"*40)
start = time.time()
print("Processing...")
try:
result = agent.execute_task(task, max_iterations=2)
elapsed = time.time() - start
print(f"\n✅ Completed in {elapsed:.2f} seconds")
if result.get('success'):
print(f"Success: True")
if result.get('final_answer'):
print(f"Answer: {result['final_answer']}")
else:
print(f"Success: False")
if result.get('error'):
print(f"Error: {result['error']}")
print(f"Iterations: {result.get('iterations', 0)}")
print(f"Tool calls: {len(result['trajectory'].tool_calls)}")
except KeyboardInterrupt:
print("\n⚠️ Interrupted")
except Exception as e:
print(f"\n❌ Error: {str(e)}")
print("="*60)
quick_test_kimi.py¶
#!/usr/bin/env python3
"""
Quick test script to verify Kimi K3 model integration
"""
import os
from dotenv import load_dotenv
from agent import ContextAwareAgent, ContextMode
# Load environment variables
load_dotenv()
def main():
# Get API key
api_key = os.getenv("MOONSHOT_API_KEY")
if not api_key:
print("❌ ERROR: MOONSHOT_API_KEY not set")
print("Please add to your .env file:")
print(" MOONSHOT_API_KEY=your_api_key_here")
return
print("🚀 Testing Kimi K3 Model (kimi-k3)")
print("=" * 50)
try:
# Create agent with Kimi provider
agent = ContextAwareAgent(
api_key=api_key,
provider="kimi",
context_mode=ContextMode.FULL,
verbose=False
)
print(f"✅ Agent created successfully")
print(f" Provider: {agent.provider}")
print(f" Model: {agent.model}")
print(f" Base URL: {agent.client.base_url}")
# Test simple query
print("\n📝 Testing basic query...")
query = "What is 2 + 2?"
response = agent.process(query)
print(f" Query: {query}")
print(f" Response: {response}")
print("\n✅ Kimi K3 integration is working!")
except Exception as e:
print(f"\n❌ Error: {e}")
if __name__ == "__main__":
main()
quickstart.py¶
#!/usr/bin/env python3
"""
Quick Start Script for Context-Aware Agent
Run this to test the agent with a simple example
"""
import os
import sys
from agent import ContextAwareAgent, ContextMode
from config import Config
def main():
"""Quick start demonstration"""
print("\n" + "="*60)
print("CONTEXT-AWARE AGENT - QUICK START")
print("="*60)
# Check for API key
api_key = os.getenv("SILICONFLOW_API_KEY")
if not api_key:
print("\n❌ ERROR: SILICONFLOW_API_KEY not found!")
print("\nPlease set your API key:")
print("1. Copy env.example to .env")
print("2. Add your API key to .env")
print("3. Or export SILICONFLOW_API_KEY=your_key_here")
sys.exit(1)
print("\n✅ API key found!")
# Simple demonstration task
demo_task = """
Please help me with the following financial calculation:
1. I have $10,000 USD that I want to convert to EUR, GBP, and JPY
2. Calculate the average amount across all three currencies (converted back to USD)
3. If I invest this average amount with a 5% annual return, what will it be worth in 2 years?
Show all your calculations step by step.
"""
print("\n📋 Demo Task:")
print("-"*40)
print(demo_task)
print("-"*40)
# Run with full context (baseline)
print("\n🚀 Running agent with FULL context...")
agent_full = ContextAwareAgent(api_key, ContextMode.FULL)
result_full = agent_full.execute_task(demo_task)
print("\n✨ Results with FULL Context:")
print(f"Success: {result_full.get('success', False)}")
print(f"Tool calls made: {len(result_full['trajectory'].tool_calls)}")
print(f"Iterations: {result_full.get('iterations', 0)}")
if result_full.get('final_answer'):
print(f"\nFinal Answer:")
print("-"*40)
print(result_full['final_answer'])
# Demonstrate context ablation effect
print("\n" + "="*60)
print("DEMONSTRATING CONTEXT ABLATION")
print("="*60)
print("\n🔬 Running same task with NO TOOL RESULTS context...")
print("(Agent won't see the results of its tool calls)")
agent_ablated = ContextAwareAgent(api_key, ContextMode.NO_TOOL_RESULTS)
result_ablated = agent_ablated.execute_task(demo_task)
print("\n⚠️ Results with NO TOOL RESULTS:")
print(f"Success: {result_ablated.get('success', False)}")
print(f"Tool calls made: {len(result_ablated['trajectory'].tool_calls)}")
print(f"Iterations: {result_ablated.get('iterations', 0)}")
if result_ablated.get('final_answer'):
print(f"\nFinal Answer (likely incorrect):")
print("-"*40)
print(result_ablated['final_answer'][:500] + "...")
# Summary
print("\n" + "="*60)
print("COMPARISON SUMMARY")
print("="*60)
print("\n📊 Key Observations:")
print(f"1. Full Context: {'✅ Success' if result_full.get('success') else '❌ Failed'}")
print(f"2. No Tool Results: {'✅ Success' if result_ablated.get('success') else '❌ Failed'}")
print(f"3. Efficiency difference: {result_ablated.get('iterations', 0) - result_full.get('iterations', 0)} more iterations without tool results")
print("\n💡 Insight:")
print("Without seeing tool results, the agent operates blind and may:")
print("- Make incorrect calculations")
print("- Repeat operations unnecessarily")
print("- Fail to validate its work")
print("\n" + "="*60)
print("Quick start complete! 🎉")
print("\nNext steps:")
print("1. Run full ablation study: python main.py --mode ablation")
print("2. Try interactive mode: python main.py --mode interactive")
print("3. Read the README.md for more details")
print("="*60 + "\n")
if __name__ == "__main__":
main()
test_agent.py¶
#!/usr/bin/env python3
"""
Test script for Context-Aware Agent
Validates installation and basic functionality
"""
import sys
import json
from agent import ContextAwareAgent, ContextMode, ToolRegistry
import unittest
from unittest.mock import MagicMock, patch
class TestToolRegistry(unittest.TestCase):
"""Test the tool registry functions"""
def test_calculator(self):
"""Test calculator tool"""
tools = ToolRegistry()
# Basic arithmetic
result = tools.calculate("2 + 2")
self.assertEqual(result["result"], 4)
# Complex expression
result = tools.calculate("(10 * 5) + (20 / 4)")
self.assertEqual(result["result"], 55.0)
# With math functions
result = tools.calculate("sqrt(16) + abs(-5)")
self.assertEqual(result["result"], 9.0)
def test_currency_converter(self):
"""Test currency conversion tool"""
tools = ToolRegistry()
# USD to EUR
result = tools.convert_currency(100, "USD", "EUR")
self.assertIn("converted_amount", result)
self.assertIn("exchange_rate", result)
self.assertGreater(result["converted_amount"], 0)
# Invalid currency
result = tools.convert_currency(100, "XXX", "YYY")
self.assertIn("error", result)
def test_pdf_parser_structure(self):
"""Test PDF parser structure (without actual PDF)"""
tools = ToolRegistry()
# Test with invalid URL (should handle gracefully)
result = tools.parse_pdf("http://invalid-url-for-testing.com/test.pdf")
self.assertIn("error", result)
class TestContextModes(unittest.TestCase):
"""Test different context modes"""
@patch.dict('os.environ', {'SILICONFLOW_API_KEY': 'test_key'})
def setUp(self):
"""Set up test fixtures"""
self.api_key = "test_key"
def test_context_mode_initialization(self):
"""Test agent initialization with different context modes"""
for mode in ContextMode:
agent = ContextAwareAgent(self.api_key, mode)
self.assertEqual(agent.context_mode, mode)
self.assertEqual(agent.trajectory.context_mode, mode)
def test_context_building(self):
"""Test context building for different modes"""
# Full context mode
agent = ContextAwareAgent(self.api_key, ContextMode.FULL)
agent.trajectory.reasoning_steps = ["Step 1", "Step 2"]
agent.trajectory.tool_calls.append(
MagicMock(tool_name="test", arguments={}, result={"test": "result"})
)
context = agent._build_context()
self.assertIn("Previous Reasoning Steps", context)
self.assertIn("Tool Call History", context)
# No reasoning mode
agent_no_reasoning = ContextAwareAgent(self.api_key, ContextMode.NO_REASONING)
agent_no_reasoning.trajectory.reasoning_steps = ["Step 1"]
context = agent_no_reasoning._build_context()
self.assertNotIn("Previous Reasoning Steps", context)
# No history mode
agent_no_history = ContextAwareAgent(self.api_key, ContextMode.NO_HISTORY)
agent_no_history.trajectory.tool_calls.append(
MagicMock(tool_name="test", arguments={}, result={"test": "result"})
)
context = agent_no_history._build_context()
self.assertEqual(context, "")
class TestAblationScenarios(unittest.TestCase):
"""Test ablation scenarios"""
def test_tool_execution(self):
"""Test tool execution"""
agent = ContextAwareAgent("test_key", ContextMode.FULL)
# Test calculator execution
result = agent._execute_tool("calculate", {"expression": "2 + 2"})
self.assertEqual(result["result"], 4)
# Test unknown tool
result = agent._execute_tool("unknown_tool", {})
self.assertIn("error", result)
def test_trajectory_reset(self):
"""Test trajectory reset"""
agent = ContextAwareAgent("test_key", ContextMode.FULL)
# Add some data to trajectory
agent.trajectory.reasoning_steps.append("Test step")
agent.trajectory.tool_calls.append(
MagicMock(tool_name="test", arguments={})
)
# Reset
agent.reset()
# Check if cleared
self.assertEqual(len(agent.trajectory.reasoning_steps), 0)
self.assertEqual(len(agent.trajectory.tool_calls), 0)
self.assertEqual(agent.trajectory.context_mode, ContextMode.FULL)
def run_integration_test():
"""Run a simple integration test"""
print("\n" + "="*60)
print("INTEGRATION TEST")
print("="*60)
# Check if API key is available
import os
api_key = os.getenv("SILICONFLOW_API_KEY")
if not api_key:
print("⚠️ Skipping integration test (no API key found)")
print("Set SILICONFLOW_API_KEY to run integration tests")
return False
print("✅ API key found, running integration test...")
try:
# Create agent
agent = ContextAwareAgent(api_key, ContextMode.FULL)
# Simple task that doesn't require external PDFs
simple_task = "Calculate: What is 15% of $2500? Then convert the result to EUR."
print(f"\nTest task: {simple_task}")
print("Running...")
# Execute with timeout
import signal
def timeout_handler(signum, frame):
raise TimeoutError("Integration test timed out")
# Set 30 second timeout
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(30)
try:
result = agent.execute_task(simple_task, max_iterations=3)
signal.alarm(0) # Cancel alarm
print(f"\n✅ Integration test completed!")
print(f"Success: {result.get('success', False)}")
print(f"Tool calls: {len(result['trajectory'].tool_calls)}")
if result.get('final_answer'):
print(f"Answer preview: {result['final_answer'][:100]}...")
return True
except TimeoutError:
print("❌ Integration test timed out")
return False
except Exception as e:
print(f"❌ Integration test failed: {str(e)}")
return False
def main():
"""Main test runner"""
print("\n" + "="*60)
print("CONTEXT-AWARE AGENT TEST SUITE")
print("="*60)
# Run unit tests
print("\n📋 Running unit tests...")
# Create test suite
loader = unittest.TestLoader()
suite = unittest.TestSuite()
# Add test cases
suite.addTests(loader.loadTestsFromTestCase(TestToolRegistry))
suite.addTests(loader.loadTestsFromTestCase(TestContextModes))
suite.addTests(loader.loadTestsFromTestCase(TestAblationScenarios))
# Run tests
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)
# Summary
print("\n" + "="*60)
print("UNIT TEST SUMMARY")
print("="*60)
print(f"Tests run: {result.testsRun}")
print(f"Failures: {len(result.failures)}")
print(f"Errors: {len(result.errors)}")
if result.wasSuccessful():
print("✅ All unit tests passed!")
else:
print("❌ Some tests failed")
sys.exit(1)
# Run integration test if possible
print("\n" + "="*60)
integration_success = run_integration_test()
# Final summary
print("\n" + "="*60)
print("FINAL TEST SUMMARY")
print("="*60)
if result.wasSuccessful():
print("✅ Unit tests: PASSED")
else:
print("❌ Unit tests: FAILED")
if integration_success:
print("✅ Integration test: PASSED")
else:
print("⚠️ Integration test: SKIPPED or FAILED")
print("\n🎉 Testing complete!")
print("="*60 + "\n")
return 0 if result.wasSuccessful() else 1
if __name__ == "__main__":
sys.exit(main())
test_code_interpreter.py¶
#!/usr/bin/env python3
"""
Test the code_interpreter tool with the agent
"""
import os
from agent import ContextAwareAgent, ContextMode
def test_code_interpreter():
"""Test code interpreter integration"""
print("\n" + "="*60)
print("🧪 CODE INTERPRETER TEST")
print("="*60)
# Check API key
api_key = os.getenv("SILICONFLOW_API_KEY")
if not api_key:
print("⚠️ No API key set, using mock test")
# Test just the tool directly
from agent import ToolRegistry
tools = ToolRegistry()
code = """
# Calculate total expenses
expenses_usd = {
'US Office': 2500000,
'UK Office (converted)': 2278481.01,
'Japan Office (converted)': 2541806.02,
'EU Office (converted)': 2282608.70,
'Singapore Office (converted)': 2388059.70
}
# Calculate total
total = sum(expenses_usd.values())
# Calculate percentages
for office, amount in expenses_usd.items():
percentage = (amount / total) * 100
print(f"{office}: ${amount:,.2f} ({percentage:.2f}%)")
print(f"\\nTotal Expenses: ${total:,.2f}")
# Calculate after 12% reduction
reduced_total = total * 0.88
savings = total - reduced_total
print(f"After 12% reduction: ${reduced_total:,.2f}")
print(f"Savings: ${savings:,.2f}")
result = {
'total': total,
'reduced': reduced_total,
'savings': savings
}
"""
result = tools.code_interpreter(code)
if result['success']:
print("✅ Code interpreter executed successfully!")
print("\nOutput:")
print(result['output'])
print(f"\nResult dictionary: {result['result']}")
else:
print(f"❌ Error: {result['error']}")
return
# Test with full agent
agent = ContextAwareAgent(api_key, ContextMode.FULL)
task = """
Calculate the following:
Given these expenses:
- US: $2,500,000
- UK: $2,278,481
- Japan: $2,541,806
- EU: $2,282,609
- Singapore: $2,388,060
Use the code_interpreter tool to:
1. Calculate the total expenses
2. Calculate what percentage each office represents
3. Calculate the new totals if we apply a 12% cost reduction
FINAL ANSWER: Provide the total, the percentage breakdown, and the reduced total.
"""
print("Running task with agent...")
print("Task: Calculate totals and percentages using code_interpreter")
print("-"*40)
result = agent.execute_task(task, max_iterations=3)
print(f"\nSuccess: {result.get('success', False)}")
print(f"Tool calls made: {len(result['trajectory'].tool_calls)}")
# Check if code_interpreter was used
code_interpreter_used = any(
tc.tool_name == 'code_interpreter'
for tc in result['trajectory'].tool_calls
)
if code_interpreter_used:
print("✅ Code interpreter was used!")
# Show the code that was executed
for tc in result['trajectory'].tool_calls:
if tc.tool_name == 'code_interpreter':
print("\nExecuted code:")
print("-"*40)
print(tc.arguments.get('code', 'N/A'))
print("-"*40)
if tc.result and tc.result.get('output'):
print("\nOutput:")
print(tc.result['output'])
else:
print("⚠️ Code interpreter was not used")
if result.get('final_answer'):
print("\n📝 Final Answer:")
print(result['final_answer'])
if __name__ == "__main__":
test_code_interpreter()
test_conversation_history.py¶
#!/usr/bin/env python3
"""
Test script to verify conversation history persistence
"""
import os
from dotenv import load_dotenv
from agent import ContextAwareAgent, ContextMode
import json
# Load environment variables
load_dotenv()
def test_conversation_history():
"""Test that conversation history persists between tasks"""
print("🧪 Testing Conversation History Persistence")
print("=" * 50)
# Get API key (use any available provider)
api_key = os.getenv("ARK_API_KEY") or os.getenv("MOONSHOT_API_KEY") or os.getenv("SILICONFLOW_API_KEY")
provider = "doubao" if os.getenv("ARK_API_KEY") else ("kimi" if os.getenv("MOONSHOT_API_KEY") else "siliconflow")
if not api_key:
print("❌ No API key found. Please set one of:")
print(" - ARK_API_KEY")
print(" - MOONSHOT_API_KEY")
print(" - SILICONFLOW_API_KEY")
return False
print(f"Using provider: {provider}")
print("-" * 50)
try:
# Create agent
agent = ContextAwareAgent(
api_key=api_key,
provider=provider,
context_mode=ContextMode.FULL,
verbose=False
)
# Test 1: First query
print("\n📝 Test 1: First query")
query1 = "Remember that my favorite number is 42. What is 10 + 5?"
result1 = agent.execute_task(query1)
print(f"Query: {query1}")
print(f"Response: {result1.get('final_answer', 'No answer')}")
# Check conversation history
print(f"\n📚 Conversation history after first query:")
print(f" Total messages: {len(agent.conversation_history)}")
# Print message roles
for i, msg in enumerate(agent.conversation_history):
role = msg.get('role', 'unknown')
content_preview = str(msg.get('content', ''))[:50] + "..." if len(str(msg.get('content', ''))) > 50 else str(msg.get('content', ''))
print(f" Message {i}: Role={role}, Content={content_preview}")
# Test 2: Second query that references first
print("\n📝 Test 2: Second query (should remember context)")
query2 = "What was my favorite number that I mentioned earlier?"
result2 = agent.execute_task(query2)
print(f"Query: {query2}")
print(f"Response: {result2.get('final_answer', 'No answer')}")
# Check if 42 is mentioned in the response
if "42" in str(result2.get('final_answer', '')):
print("✅ SUCCESS: Agent remembered the favorite number from conversation history!")
else:
print("⚠️ WARNING: Agent might not have remembered the number. Check response above.")
# Check conversation history growth
print(f"\n📚 Conversation history after second query:")
print(f" Total messages: {len(agent.conversation_history)}")
# Test 3: Verify system prompt unchanged
print("\n📝 Test 3: Verify system prompt remains unchanged")
system_prompt = agent.conversation_history[0].get('content', '')
if "favorite number" not in system_prompt and "42" not in system_prompt:
print("✅ SUCCESS: System prompt remains unchanged!")
else:
print("❌ FAILURE: System prompt was modified!")
# Test 4: Reset and verify history cleared
print("\n📝 Test 4: Test reset functionality")
agent.reset()
print(f" Messages after reset: {len(agent.conversation_history)}")
if len(agent.conversation_history) == 1 and agent.conversation_history[0]['role'] == 'system':
print("✅ SUCCESS: Reset properly cleared history and kept system prompt!")
else:
print("❌ FAILURE: Reset did not work correctly!")
# Test 5: New conversation after reset
print("\n📝 Test 5: New conversation after reset")
query3 = "What was my favorite number?"
result3 = agent.execute_task(query3)
print(f"Query: {query3}")
print(f"Response: {result3.get('final_answer', 'No answer')}")
if "42" not in str(result3.get('final_answer', '')) and "don't" in str(result3.get('final_answer', '').lower()):
print("✅ SUCCESS: Agent correctly doesn't remember after reset!")
else:
print("⚠️ Check if agent properly forgot the previous conversation")
print("\n" + "=" * 50)
print("Conversation history tests complete!")
return True
except Exception as e:
print(f"\n❌ Error during test: {e}")
import traceback
traceback.print_exc()
return False
if __name__ == "__main__":
success = test_conversation_history()
exit(0 if success else 1)
test_deepseek.py¶
#!/usr/bin/env python3
"""
Test script for DeepSeek model integration.
Tests deepseek-v4-flash (default) with conversation and tool calling.
"""
import os
import sys
from dotenv import load_dotenv
from agent import ContextAwareAgent, ContextMode
from config import Config
# Load environment variables
load_dotenv()
def test_basic_conversation():
"""Test basic conversation capabilities"""
print("\n" + "=" * 60)
print("TEST 1: Basic Conversation")
print("=" * 60)
try:
api_key = os.getenv("DEEPSEEK_API_KEY")
if not api_key:
print("❌ ERROR: DEEPSEEK_API_KEY not set in environment")
print("Please set it in your .env file or as environment variable")
return False
agent = ContextAwareAgent(
api_key=api_key,
provider="deepseek",
context_mode=ContextMode.FULL,
verbose=False,
)
query = "What is 25 * 4 + 10? Reply with FINAL ANSWER: and the number."
print(f"\n📝 Query: {query}")
response = agent.process(query)
print(f"\n🤖 Response: {response}")
if "110" in response:
print("\n✅ Basic conversation test passed!")
return True
else:
print("\n❌ Test failed - incorrect answer")
return False
except Exception as e:
print(f"\n❌ Error during test: {e}")
return False
def test_tool_usage():
"""Test tool calling capabilities"""
print("\n" + "=" * 60)
print("TEST 2: Tool Usage (Calculator)")
print("=" * 60)
try:
api_key = os.getenv("DEEPSEEK_API_KEY")
if not api_key:
print("❌ ERROR: DEEPSEEK_API_KEY not set")
return False
agent = ContextAwareAgent(
api_key=api_key,
provider="deepseek",
context_mode=ContextMode.FULL,
verbose=False,
)
query = (
"Calculate: (123.45 * 67.89) / 12.34 + sqrt(144) - 2^8. "
"Use the calculate tool. End with FINAL ANSWER:"
)
print(f"\n📝 Query: {query}")
response = agent.process(query)
print(f"\n🤖 Response: {response}")
if agent.trajectory.tool_calls:
print(f"\n🔧 Tools used: {len(agent.trajectory.tool_calls)}")
for call in agent.trajectory.tool_calls:
print(f" - {call.tool_name}: {call.arguments}")
print("\n✅ Tool usage test passed!")
return True
else:
print("\n⚠️ No tools were used")
return False
except Exception as e:
print(f"\n❌ Error during test: {e}")
return False
def test_currency_conversion():
"""Test currency conversion tool"""
print("\n" + "=" * 60)
print("TEST 3: Currency Conversion")
print("=" * 60)
try:
api_key = os.getenv("DEEPSEEK_API_KEY")
if not api_key:
print("❌ ERROR: DEEPSEEK_API_KEY not set")
return False
agent = ContextAwareAgent(
api_key=api_key,
provider="deepseek",
context_mode=ContextMode.FULL,
verbose=False,
)
query = "Convert 100 USD to EUR and JPY. Use convert_currency. FINAL ANSWER: the amounts."
print(f"\n📝 Query: {query}")
response = agent.process(query)
print(f"\n🤖 Response: {response}")
tool_names = [call.tool_name for call in agent.trajectory.tool_calls]
if "convert_currency" in tool_names:
print("\n🔧 Currency converter was used")
print("\n✅ Currency conversion test passed!")
return True
else:
print("\n⚠️ Currency converter was not used")
return False
except Exception as e:
print(f"\n❌ Error during test: {e}")
return False
def test_model_info():
"""Test and display model information"""
print("\n" + "=" * 60)
print("TEST 4: Model Information")
print("=" * 60)
try:
api_key = os.getenv("DEEPSEEK_API_KEY")
if not api_key:
print("❌ ERROR: DEEPSEEK_API_KEY not set")
return False
agent = ContextAwareAgent(
api_key=api_key,
provider="deepseek",
context_mode=ContextMode.FULL,
verbose=False,
)
expected = Config.get_default_model("deepseek")
print("\n📊 Model Configuration:")
print(f" Provider: {agent.provider}")
print(f" Model: {agent.model}")
print(f" Expected default: {expected}")
print(f" Base URL: {agent.client.base_url}")
print(f" Context Mode: {agent.context_mode.value}")
if agent.provider != "deepseek" or agent.model != expected:
print("\n❌ Model config mismatch")
return False
print("\n✅ Model info test completed!")
return True
except Exception as e:
print(f"\n❌ Error during test: {e}")
return False
def main():
"""Run all tests"""
print("\n" + "=" * 60)
print("DEEPSEEK MODEL INTEGRATION TEST SUITE")
print("=" * 60)
print("\nModel: deepseek-v4-flash (default)")
print("Provider: DeepSeek")
print("API: https://api.deepseek.com")
if not os.getenv("DEEPSEEK_API_KEY"):
print("\n❌ ERROR: DEEPSEEK_API_KEY not found in environment")
print("\nPlease set up your .env file with:")
print(" DEEPSEEK_API_KEY=your_api_key_here")
print("\nYou can get an API key from: https://platform.deepseek.com/api_keys")
sys.exit(1)
results = []
results.append(("Model Information", test_model_info()))
results.append(("Basic Conversation", test_basic_conversation()))
results.append(("Tool Usage", test_tool_usage()))
results.append(("Currency Conversion", test_currency_conversion()))
print("\n" + "=" * 60)
print("TEST SUMMARY")
print("=" * 60)
passed = sum(1 for _, result in results if result)
total = len(results)
for test_name, result in results:
status = "✅ PASSED" if result else "❌ FAILED"
print(f" {test_name}: {status}")
print(f"\nTotal: {passed}/{total} tests passed")
if passed == total:
print("\n🎉 All tests passed! DeepSeek integration is working correctly.")
else:
print(f"\n⚠️ {total - passed} test(s) failed. Please check the errors above.")
return passed == total
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)
test_default_provider.py¶
#!/usr/bin/env python3
"""
Test that Doubao is the default provider
"""
import os
import sys
# Test without any arguments - should use Doubao
print("Testing default provider...")
# Check if ARK_API_KEY is available
ark_key = os.getenv("ARK_API_KEY")
sf_key = os.getenv("SILICONFLOW_API_KEY")
print(f"ARK_API_KEY available: {'Yes' if ark_key else 'No'}")
print(f"SILICONFLOW_API_KEY available: {'Yes' if sf_key else 'No'}")
if ark_key:
from agent import ContextAwareAgent, ContextMode
from config import Config
# Check config default
print(f"\nConfig default provider: {Config.LLM_PROVIDER}")
# Create agent with default provider from config
agent = ContextAwareAgent(ark_key, ContextMode.FULL, provider=Config.LLM_PROVIDER)
print(f"\n✅ Default agent created successfully!")
print(f"Provider: {agent.provider}")
print(f"Model: {agent.model}")
print(f"Base URL: {agent.client.base_url}")
if agent.provider == "doubao":
print("\n🎉 SUCCESS: Doubao is the default provider!")
else:
print(f"\n❌ ERROR: Expected doubao, got {agent.provider}")
sys.exit(1)
else:
print("\n⚠️ ARK_API_KEY not set. Cannot test default provider.")
print("Please set: export ARK_API_KEY=your_key_here")
test_doubao.py¶
#!/usr/bin/env python3
"""
Quick test for Doubao provider
"""
import os
from agent import ContextAwareAgent, ContextMode
def test_doubao():
"""Test Doubao provider with a simple task"""
print("\n" + "="*60)
print("🧪 DOUBAO PROVIDER TEST")
print("="*60)
# Check for API key
api_key = os.getenv("ARK_API_KEY")
if not api_key:
print("❌ ARK_API_KEY not found. Please set it to test Doubao provider.")
print(" export ARK_API_KEY=your_key_here")
return
print("✅ ARK API key found")
# Create agent with Doubao provider
try:
agent = ContextAwareAgent(api_key, ContextMode.FULL, provider="doubao")
print(f"✅ Agent created with Doubao provider")
print(f" Model: {agent.model}")
print(f" Base URL: {agent.client.base_url}")
# Simple test task (minimal to save tokens)
print("\n📝 Running simple test task...")
task = "Calculate: What is 15 + 27? Provide FINAL ANSWER with the result."
result = agent.execute_task(task, max_iterations=3)
if result.get('success'):
print("✅ Task executed successfully!")
if result.get('final_answer'):
print(f" Answer: {result['final_answer'][:100]}...")
else:
print(f"⚠️ Task did not complete successfully")
if result.get('error'):
print(f" Error: {result['error']}")
print(f"\n📊 Execution stats:")
print(f" Iterations: {result.get('iterations', 0)}")
print(f" Tool calls: {len(result['trajectory'].tool_calls)}")
except Exception as e:
print(f"❌ Error: {str(e)}")
print("\nNote: Make sure your ARK_API_KEY is valid and has access to the doubao model.")
print("\n" + "="*60)
if __name__ == "__main__":
test_doubao()
test_kimi.py¶
#!/usr/bin/env python3
"""
Test script for Kimi K3 model integration
Tests the Kimi K3 model (kimi-k3) with various tasks
"""
import os
import sys
from dotenv import load_dotenv
from agent import ContextAwareAgent, ContextMode
from config import Config
# Load environment variables
load_dotenv()
def test_basic_conversation():
"""Test basic conversation capabilities"""
print("\n" + "="*60)
print("TEST 1: Basic Conversation")
print("="*60)
try:
# Get API key
api_key = os.getenv("MOONSHOT_API_KEY")
if not api_key:
print("❌ ERROR: MOONSHOT_API_KEY not set in environment")
print("Please set it in your .env file or as environment variable")
return False
# Create agent
agent = ContextAwareAgent(
api_key=api_key,
provider="kimi",
context_mode=ContextMode.FULL,
verbose=False
)
# Test basic conversation
query = "What is 25 * 4 + 10?"
print(f"\n📝 Query: {query}")
response = agent.process(query)
print(f"\n🤖 Response: {response}")
# Verify response contains correct answer
if "110" in response:
print("\n✅ Basic conversation test passed!")
return True
else:
print("\n❌ Test failed - incorrect answer")
return False
except Exception as e:
print(f"\n❌ Error during test: {e}")
return False
def test_tool_usage():
"""Test tool calling capabilities"""
print("\n" + "="*60)
print("TEST 2: Tool Usage (Calculator)")
print("="*60)
try:
# Get API key
api_key = os.getenv("MOONSHOT_API_KEY")
if not api_key:
print("❌ ERROR: MOONSHOT_API_KEY not set")
return False
# Create agent
agent = ContextAwareAgent(
api_key=api_key,
provider="kimi",
context_mode=ContextMode.FULL,
verbose=False
)
# Test complex calculation requiring calculator tool
query = "Calculate: (123.45 * 67.89) / 12.34 + sqrt(144) - 2^8"
print(f"\n📝 Query: {query}")
response = agent.process(query)
print(f"\n🤖 Response: {response}")
# Check if calculator was used
if agent.trajectory.tool_calls:
print(f"\n🔧 Tools used: {len(agent.trajectory.tool_calls)}")
for call in agent.trajectory.tool_calls:
print(f" - {call.tool_name}: {call.arguments}")
print("\n✅ Tool usage test passed!")
return True
else:
print("\n⚠️ No tools were used")
return False
except Exception as e:
print(f"\n❌ Error during test: {e}")
return False
def test_currency_conversion():
"""Test currency conversion tool"""
print("\n" + "="*60)
print("TEST 3: Currency Conversion")
print("="*60)
try:
# Get API key
api_key = os.getenv("MOONSHOT_API_KEY")
if not api_key:
print("❌ ERROR: MOONSHOT_API_KEY not set")
return False
# Create agent
agent = ContextAwareAgent(
api_key=api_key,
provider="kimi",
context_mode=ContextMode.FULL,
verbose=False
)
# Test currency conversion
query = "Convert 100 USD to EUR and JPY"
print(f"\n📝 Query: {query}")
response = agent.process(query)
print(f"\n🤖 Response: {response}")
# Check if currency converter was used
tool_names = [call.tool_name for call in agent.trajectory.tool_calls]
if "convert_currency" in tool_names:
print(f"\n🔧 Currency converter was used")
print("\n✅ Currency conversion test passed!")
return True
else:
print("\n⚠️ Currency converter was not used")
return False
except Exception as e:
print(f"\n❌ Error during test: {e}")
return False
def test_model_info():
"""Test and display model information"""
print("\n" + "="*60)
print("TEST 4: Model Information")
print("="*60)
try:
# Get API key
api_key = os.getenv("MOONSHOT_API_KEY")
if not api_key:
print("❌ ERROR: MOONSHOT_API_KEY not set")
return False
# Create agent
agent = ContextAwareAgent(
api_key=api_key,
provider="kimi",
context_mode=ContextMode.FULL,
verbose=False
)
print(f"\n📊 Model Configuration:")
print(f" Provider: {agent.provider}")
print(f" Model: {agent.model}")
print(f" Base URL: {agent.client.base_url}")
print(f" Context Mode: {agent.context_mode.value}")
# Test model identification
query = "What model are you?"
print(f"\n📝 Query: {query}")
response = agent.process(query)
print(f"\n🤖 Response: {response}")
print("\n✅ Model info test completed!")
return True
except Exception as e:
print(f"\n❌ Error during test: {e}")
return False
def main():
"""Run all tests"""
print("\n" + "="*60)
print("KIMI K3 MODEL INTEGRATION TEST SUITE")
print("="*60)
print("\nModel: kimi-k3")
print("Provider: Moonshot AI")
print("API: https://api.moonshot.cn/v1")
# Check environment
if not os.getenv("MOONSHOT_API_KEY"):
print("\n❌ ERROR: MOONSHOT_API_KEY not found in environment")
print("\nPlease set up your .env file with:")
print(" MOONSHOT_API_KEY=your_api_key_here")
print("\nYou can get an API key from: https://platform.moonshot.cn/")
sys.exit(1)
# Run tests
results = []
# Test 1: Basic conversation
results.append(("Basic Conversation", test_basic_conversation()))
# Test 2: Tool usage
results.append(("Tool Usage", test_tool_usage()))
# Test 3: Currency conversion
results.append(("Currency Conversion", test_currency_conversion()))
# Test 4: Model information
results.append(("Model Information", test_model_info()))
# Summary
print("\n" + "="*60)
print("TEST SUMMARY")
print("="*60)
passed = sum(1 for _, result in results if result)
total = len(results)
for test_name, result in results:
status = "✅ PASSED" if result else "❌ FAILED"
print(f" {test_name}: {status}")
print(f"\nTotal: {passed}/{total} tests passed")
if passed == total:
print("\n🎉 All tests passed! Kimi K3 integration is working correctly.")
else:
print(f"\n⚠️ {total - passed} test(s) failed. Please check the errors above.")
return passed == total
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)
test_pdf_task.py¶
#!/usr/bin/env python3
"""
Test script to verify PDF parsing and currency conversion
"""
import os
import sys
from agent import ContextAwareAgent, ContextMode
def test_pdf_with_currencies():
"""Test PDF parsing with currency conversion"""
print("\n" + "="*60)
print("🧪 PDF PARSING & CURRENCY CONVERSION TEST")
print("="*60)
# Check API key
api_key = os.getenv("SILICONFLOW_API_KEY")
if not api_key:
print("❌ No API key found. Set SILICONFLOW_API_KEY environment variable.")
return False
# Create agent
agent = ContextAwareAgent(api_key, ContextMode.FULL)
# Test task
task = """
Analyze the expense report at test_pdfs/simple_expense_report.pdf
Extract the following expenses mentioned in the document:
- US Office: $2,500,000 USD
- UK Office: £1,800,000 GBP
- Japan Office: ¥380,000,000 JPY
- EU Office: €2,100,000 EUR
- Singapore Office: S$3,200,000 SGD
Convert all amounts to USD and calculate the total.
FINAL ANSWER: Provide the total expenses in USD.
"""
print("📋 Task: Parse PDF and convert multiple currencies to USD")
print("-"*40)
try:
# Execute task
result = agent.execute_task(task, max_iterations=5)
print("\n" + "="*40)
print("RESULTS:")
print("="*40)
print(f"Success: {result.get('success', False)}")
print(f"Iterations: {result.get('iterations', 0)}")
print(f"Tool Calls: {len(result['trajectory'].tool_calls)}")
# Show tool calls made
print("\n📊 Tool Calls Made:")
for i, tc in enumerate(result['trajectory'].tool_calls, 1):
print(f"{i}. {tc.tool_name}")
if tc.tool_name == "parse_pdf":
print(f" - PDF: {tc.arguments.get('url', 'N/A')}")
if tc.result and 'num_pages' in tc.result:
print(f" - Pages: {tc.result['num_pages']}")
elif tc.tool_name == "convert_currency":
print(f" - {tc.arguments.get('amount', 0)} {tc.arguments.get('from_currency', '')} → {tc.arguments.get('to_currency', '')}")
if tc.result and 'converted_amount' in tc.result:
print(f" - Result: {tc.result['converted_amount']}")
elif tc.tool_name == "calculate":
print(f" - Expression: {tc.arguments.get('expression', '')}")
if tc.result and 'result' in tc.result:
print(f" - Result: {tc.result['result']}")
if result.get('final_answer'):
print("\n✅ Final Answer:")
print("-"*40)
print(result['final_answer'])
if result.get('error'):
print(f"\n❌ Error: {result['error']}")
return result.get('success', False)
except Exception as e:
print(f"\n❌ Exception: {str(e)}")
return False
if __name__ == "__main__":
# Ensure PDFs exist
if not os.path.exists("test_pdfs/simple_expense_report.pdf"):
print("⚠️ Creating sample PDFs...")
os.system("python create_sample_pdf.py")
# Run test
success = test_pdf_with_currencies()
sys.exit(0 if success else 1)
test_provider.py¶
#!/usr/bin/env python3
"""
Test script to verify provider configuration
"""
import os
from agent import ContextAwareAgent, ContextMode
def test_providers():
"""Test different provider configurations"""
print("\n" + "="*60)
print("🧪 PROVIDER CONFIGURATION TEST")
print("="*60)
# Test SiliconFlow
sf_key = os.getenv("SILICONFLOW_API_KEY")
if sf_key:
print("\n✅ SiliconFlow API key found")
try:
agent = ContextAwareAgent(sf_key, ContextMode.FULL, provider="siliconflow")
print(f" Provider: {agent.provider}")
print(f" Model: {agent.model}")
print(f" Base URL: {agent.client.base_url}")
except Exception as e:
print(f" ❌ Error: {str(e)}")
else:
print("\n⚠️ SiliconFlow API key not found (SILICONFLOW_API_KEY)")
# Test Doubao
ark_key = os.getenv("ARK_API_KEY")
if ark_key:
print("\n✅ Doubao/ARK API key found")
try:
agent = ContextAwareAgent(ark_key, ContextMode.FULL, provider="doubao")
print(f" Provider: {agent.provider}")
print(f" Model: {agent.model}")
print(f" Base URL: {agent.client.base_url}")
except Exception as e:
print(f" ❌ Error: {str(e)}")
else:
print("\n⚠️ Doubao/ARK API key not found (ARK_API_KEY)")
# Test DeepSeek
deepseek_key = os.getenv("DEEPSEEK_API_KEY")
if deepseek_key:
print("\n✅ DeepSeek API key found")
try:
agent = ContextAwareAgent(deepseek_key, ContextMode.FULL, provider="deepseek")
print(f" Provider: {agent.provider}")
print(f" Model: {agent.model}")
print(f" Base URL: {agent.client.base_url}")
except Exception as e:
print(f" ❌ Error: {str(e)}")
else:
print("\n⚠️ DeepSeek API key not found (DEEPSEEK_API_KEY)")
# Test custom model
if sf_key:
print("\n🔧 Testing custom model specification:")
try:
agent = ContextAwareAgent(sf_key, ContextMode.FULL,
provider="siliconflow",
model="Qwen/QwQ-32B")
print(f" Provider: {agent.provider}")
print(f" Custom Model: {agent.model}")
except Exception as e:
print(f" ❌ Error: {str(e)}")
print("\n" + "="*60)
print("Test complete!")
# Show usage examples
print("\n📖 Usage Examples:")
print("-"*40)
if sf_key:
print("\n# Using SiliconFlow:")
print("python main.py --provider siliconflow")
print("python main.py --provider siliconflow --model Qwen/QwQ-32B")
if ark_key:
print("\n# Using Doubao:")
print("python main.py --provider doubao")
print("python main.py --provider doubao --model doubao-seed-1-6-thinking-250715")
if deepseek_key:
print("\n# Using DeepSeek:")
print("python main.py --provider deepseek")
print("python main.py --provider deepseek --model deepseek-v4-pro")
if not sf_key and not ark_key and not deepseek_key:
print("\n⚠️ No API keys found. Please set one of:")
print(" export SILICONFLOW_API_KEY=your_key")
print(" export ARK_API_KEY=your_key")
print(" export DEEPSEEK_API_KEY=your_key")
print("\n" + "="*60)
if __name__ == "__main__":
test_providers()
test_provider_switching.py¶
#!/usr/bin/env python3
"""
Test script to verify provider switching functionality
"""
import os
from dotenv import load_dotenv
from agent import ContextAwareAgent, ContextMode
from config import Config
# Load environment variables
load_dotenv()
def test_provider_switching():
"""Test switching between different providers"""
print("🧪 Testing Provider Switching")
print("=" * 50)
providers_to_test = []
# Check which providers have API keys configured
if os.getenv("SILICONFLOW_API_KEY"):
providers_to_test.append(("siliconflow", os.getenv("SILICONFLOW_API_KEY")))
print("✅ SiliconFlow API key found")
else:
print("⏭️ Skipping SiliconFlow (no API key)")
if os.getenv("ARK_API_KEY"):
providers_to_test.append(("doubao", os.getenv("ARK_API_KEY")))
print("✅ Doubao API key found")
else:
print("⏭️ Skipping Doubao (no API key)")
if os.getenv("MOONSHOT_API_KEY"):
providers_to_test.append(("kimi", os.getenv("MOONSHOT_API_KEY")))
print("✅ Kimi API key found")
else:
print("⏭️ Skipping Kimi (no API key)")
if os.getenv("DEEPSEEK_API_KEY"):
providers_to_test.append(("deepseek", os.getenv("DEEPSEEK_API_KEY")))
print("✅ DeepSeek API key found")
else:
print("⏭️ Skipping DeepSeek (no API key)")
if not providers_to_test:
print("\n❌ No API keys configured. Please set at least one:")
print(" - SILICONFLOW_API_KEY")
print(" - ARK_API_KEY")
print(" - MOONSHOT_API_KEY")
print(" - DEEPSEEK_API_KEY")
return
print(f"\nTesting {len(providers_to_test)} provider(s)...")
print("-" * 50)
# Test each available provider
for provider_name, api_key in providers_to_test:
print(f"\n📌 Testing {provider_name.upper()}")
try:
# Create agent with provider
agent = ContextAwareAgent(
api_key=api_key,
provider=provider_name,
context_mode=ContextMode.FULL,
verbose=False
)
# Get default model from config
default_model = Config.get_default_model(provider_name)
print(f" Provider: {agent.provider}")
print(f" Model: {agent.model}")
print(f" Expected: {default_model}")
print(f" Base URL: {agent.client.base_url}")
# Test with a simple query
query = "What is 5 + 3?"
print(f" Testing query: {query}")
response = agent.process(query)
if "8" in response:
print(f" ✅ {provider_name} working correctly!")
else:
print(f" ⚠️ {provider_name} response didn't contain expected answer")
print(f" Response: {response[:100]}...")
except Exception as e:
print(f" ❌ Error with {provider_name}: {e}")
print("\n" + "=" * 50)
print("Provider switching test complete!")
# Show summary
print("\n📊 Summary:")
print(f" Providers tested: {len(providers_to_test)}")
print(" Available providers: siliconflow, doubao, kimi, moonshot, deepseek")
if len(providers_to_test) < 3:
print("\n💡 Tip: Configure more API keys to test all providers")
if __name__ == "__main__":
test_provider_switching()
test_simple.py¶
#!/usr/bin/env python3
"""
Test with a simpler task to diagnose the issue
"""
import os
import sys
import time
from agent import ContextAwareAgent, ContextMode
def test_simple_task():
"""Test with a very simple task to check if the agent is working"""
print("\n" + "="*60)
print("🧪 SIMPLE TASK TEST")
print("="*60)
# Get API key
api_key = os.getenv("SILICONFLOW_API_KEY")
if not api_key:
print("❌ SILICONFLOW_API_KEY not found")
return
print("✅ API key found")
# Create agent
agent = ContextAwareAgent(api_key, ContextMode.FULL, provider="siliconflow")
print(f"✅ Agent created")
print(f" Model: {agent.model}")
# Very simple task - no tools needed
print("\n📝 Test 1: Simple question (no tools)")
task1 = "What is 2 + 2? Just tell me the answer. FINAL ANSWER: provide the result."
start = time.time()
print("Executing...")
try:
result = agent.execute_task(task1, max_iterations=1)
elapsed = time.time() - start
print(f"✅ Completed in {elapsed:.2f} seconds")
if result.get('final_answer'):
print(f" Answer: {result['final_answer'][:100]}")
print(f" Tool calls: {len(result['trajectory'].tool_calls)}")
except Exception as e:
print(f"❌ Error: {str(e)}")
return
# Task with a single tool
print("\n📝 Test 2: Simple calculation (with tool)")
task2 = "Use the calculate tool to compute 15 * 3. FINAL ANSWER: provide the result."
start = time.time()
print("Executing...")
try:
result = agent.execute_task(task2, max_iterations=2)
elapsed = time.time() - start
print(f"✅ Completed in {elapsed:.2f} seconds")
if result.get('final_answer'):
print(f" Answer: {result['final_answer'][:100]}")
print(f" Tool calls: {len(result['trajectory'].tool_calls)}")
except KeyboardInterrupt:
print("\n⚠️ Interrupted by user")
print("The model might be taking too long to respond.")
print("\nSuggestions:")
print("1. Try using --provider doubao for faster responses")
print("2. Check your internet connection")
print("3. The model might be overloaded - try again later")
except Exception as e:
print(f"❌ Error: {str(e)}")
print("\n" + "="*60)
if __name__ == "__main__":
test_simple_task()