单智能体架构

from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional

@dataclass
class AgentState:
    """Agent 在任一时刻的完整状态"""
    task: str                          # 原始任务描述
    messages: List[Dict] = field(default_factory=list)  # 对话历史
    plan: List[str] = field(default_factory=list)       # 当前计划
    current_step: int = 0              # 当前执行到第几步
    tool_results: List[Dict] = field(default_factory=list)  # 工具调用结果
    reflections: List[str] = field(default_factory=list)    # 反思记录
    final_answer: Optional[str] = None # 最终答案
    status: str = "running"            # running / success / failed
    retry_count: int = 0               # 重试次数

你是一个专业的数据分析助手。

## 身份
- 名称: DataAnalyst
- 专长: SQL 查询、数据可视化、统计分析
- 风格: 严谨、专业，给出结论时附带数据依据

## 可用工具
- query_database: 执行 SQL 查询
- create_chart: 生成数据图表
- run_python: 执行 Python 代码

## 行为准则
1. 先理解用户的分析需求，再选择数据源
2. 查询数据前先确认数据范围和口径
3. 给出结论时，必须附带关键数据支撑
4. 如果数据不足以得出结论，明确告知用户
5. 涉及敏感数据时，进行脱敏处理

## 输出格式
- 数据分析结论使用结构化格式
- 重要数字加粗标注
- 必要时给出可视化建议

def react_agent(task: str, tools: ToolRegistry, max_steps: int = 10):
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": task}
    ]
    
    for step in range(max_steps):
        response = llm.chat(messages, tools=tools.get_tool_descriptions())
        
        if response.has_tool_call():
            # 执行工具
            for tool_call in response.tool_calls:
                result = tools.execute(tool_call.name, tool_call.params)
                messages.append({"role": "tool", "content": str(result), 
                                 "tool_call_id": tool_call.id})
        else:
            # LLM 直接给出最终回答
            return response.content
    
    return "达到最大步数限制，未能完成任务"

def plan_and_execute(task: str, tools: ToolRegistry):
    # 阶段 1: 规划
    plan = llm.generate(f"""
        任务: {task}
        可用工具: {tools.get_tool_names()}
        请制定一个详细的执行计划，每步包含：要做什么、用哪个工具。
    """)
    
    steps = parse_plan(plan)  # 解析为步骤列表
    
    # 阶段 2: 逐步执行
    results = []
    for i, step in enumerate(steps):
        # 每步执行前，考虑之前的结果
        context = f"已完成步骤: {results}\n当前步骤: {step}"
        
        action = llm.generate(f"根据上下文执行当前步骤: {context}")
        
        if action.has_tool_call():
            result = tools.execute(action.tool_call.name, action.tool_call.params)
            results.append({"step": i, "action": action, "result": result})
        else:
            results.append({"step": i, "action": action, "result": action.content})
    
    # 阶段 3: 汇总结果
    final = llm.generate(f"任务: {task}\n执行结果: {results}\n请给出最终回答。")
    return final

class ContextManager:
    """管理 Agent 的上下文窗口"""
    
    def __init__(self, max_tokens=8000):
        self.max_tokens = max_tokens
        self.system_prompt = ""
        self.messages = []
        self.tool_results = []
    
    def add_message(self, message):
        self.messages.append(message)
        self._trim_if_needed()
    
    def _trim_if_needed(self):
        """当上下文超过限制时进行裁剪"""
        total = count_tokens(self.system_prompt + self.messages + self.tool_results)
        
        if total <= self.max_tokens:
            return
        
        # 策略 1: 压缩旧的工具结果
        old_results = self.tool_results[:len(self.tool_results)//2]
        if old_results:
            summary = llm.summarize(old_results)
            self.tool_results = [summary] + self.tool_results[len(old_results):]
        
        # 策略 2: 裁剪早期对话
        while count_tokens(self.messages) > self.max_tokens * 0.6:
            self.messages.pop(0)  # 移除最早的消息

适合单 Agent:
✓ 任务相对单一，领域集中
✓ 步骤之间有强依赖关系
✓ 不需要并行处理
✓ 交互性强，需要持续对话

适合多 Agent:
✓ 任务涉及多个不同领域
✓ 子任务之间相互独立，可并行
✓ 需要多角色协作（如审稿人、作者、编辑）
✓ 单个 Agent 上下文装不下所有信息