Agent 设计模式与工程化最佳实践

┌─────────────────────────────────────────────────────────────┐
│                    ReAct 执行循环                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  输入：任务描述                                              │
│         ↓                                                   │
│  ┌─────────────┐                                           │
│  │  Thought    │  分析当前状态，推理下一步行动              │
│  │  (推理)     │  示例："我需要先搜索相关信息"              │
│  └──────┬──────┘                                           │
│         │                                                   │
│         ↓                                                   │
│  ┌─────────────┐                                           │
│  │   Action    │  执行具体行动（调用工具/API）              │
│  │  (行动)     │  示例：Search("量子计算最新进展")          │
│  └──────┬──────┘                                           │
│         │                                                   │
│         ↓                                                   │
│  ┌─────────────┐                                           │
│  │ Observation │  获取行动结果                              │
│  │  (观察)     │  示例："2026 年 3 月最新论文显示..."        │
│  └──────┬──────┘                                           │
│         │                                                   │
│         ↓                                                   │
│  ┌─────────────┐                                           │
│  │   判断      │  任务是否完成？                            │
│  └──────┬──────┘                                           │
│         │                                                   │
│    ┌────┴────┐                                             │
│    │         │                                             │
│   是        否                                             │
│    │         │                                             │
│    ↓         └──────────┐                                  │
│  ┌─────────────┐       │                                  │
│  │ Final Answer│       │                                  │
│  │ (最终答案)  │       │                                  │
│  └─────────────┘       │                                  │
│                        │                                  │
│                        └──────────→ 返回 Thought           │
│                                                             │
└─────────────────────────────────────────────────────────────┘

关键特性：
  • 推理透明：每一步思考都记录在案
  • 行动可解释：行动基于明确的推理
  • 支持多步：可以执行任意次数的循环
  • 易于调试：完整的执行轨迹
                        

class ReActAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
    
    def run(self, task, max_iterations=10):
        """执行 ReAct 循环"""
        history = []
        
        for i in range(max_iterations):
            # Step 1: Thought (推理)
            thought = self._generate_thought(task, history)
            history.append(f"Thought {i+1}: {thought}")
            
            # Step 2: Action (行动)
            action, action_input = self._parse_action(thought)
            
            if action == "Final Answer":
                return action_input
            
            history.append(f"Action {i+1}: {action}({action_input})")
            
            # Step 3: Observation (观察)
            observation = self._execute_action(action, action_input)
            history.append(f"Observation {i+1}: {observation}")
        
        raise Exception("达到最大迭代次数")
    
    def _generate_thought(self, task, history):
        """生成推理步骤"""
        prompt = f"""
任务：{task}
历史：
{chr(10).join(history)}

请思考下一步应该做什么。如果需要调用工具，请使用以下格式：
Action: tool_name
Action Input: query

如果已有足够信息回答问题，请使用：
Final Answer: 你的答案

Thought: """
        
        response = self.llm.generate(prompt)
        return response
    
    def _parse_action(self, thought):
        """解析行动"""
        if "Final Answer:" in thought:
            return "Final Answer", thought.split("Final Answer:")[1].strip()
        
        action_line = [l for l in thought.split('\n') if l.startswith("Action:")][0]
        action = action_line.replace("Action:", "").strip()
        
        input_line = [l for l in thought.split('\n') if l.startswith("Action Input:")][0]
        action_input = input_line.replace("Action Input:", "").strip()
        
        return action, action_input
    
    def _execute_action(self, action, action_input):
        """执行行动"""
        if action not in self.tools:
            return f"错误：未知工具 {action}"
        
        try:
            result = self.tools[action].run(action_input)
            return str(result)
        except Exception as e:
            return f"错误：{str(e)}"

# 使用示例
tools = [SearchTool(), CalculatorTool(), CodeExecutorTool()]
agent = ReActAgent(llm=gpt4, tools=tools)

result = agent.run("2026 年诺贝尔物理学奖得主是谁？他们的研究领域是什么？")
print(result)
                        

class ContextManager:
    def __init__(self, max_tokens=128000):
        self.max_tokens = max_tokens
        self.context = []
    
    def add(self, content, priority="normal"):
        """添加内容到上下文"""
        self.context.append({
            "content": content,
            "priority": priority,
            "tokens": self._count_tokens(content)
        })
        self._trim_if_needed()
    
    def _trim_if_needed(self):
        """根据优先级截断上下文"""
        total_tokens = sum(item["tokens"] for item in self.context)
        
        if total_tokens > self.max_tokens:
            # 按优先级排序：system > high > normal > low
            priority_order = {"system": 0, "high": 1, "normal": 2, "low": 3}
            self.context.sort(key=lambda x: priority_order[x["priority"]])
            
            # 从低优先级开始移除
            while total_tokens > self.max_tokens and self.context:
                if self.context[-1]["priority"] == "system":
                    break  # 不移除系统提示
                removed = self.context.pop()
                total_tokens -= removed["tokens"]
    
    def get_context(self):
        """获取当前上下文"""
        return "\n".join(item["content"] for item in self.context)
    
    def _count_tokens(self, text):
        """估算 token 数（简化版）"""
        return len(text) // 4  # 粗略估算：1 token ≈ 4 字符

# 使用示例
manager = ContextManager(max_tokens=8000)

# 添加系统提示（最高优先级）
manager.add("你是一个专业的研究助手。", priority="system")

# 添加任务描述（高优先级）
manager.add("任务：分析 2026 年 AI 趋势", priority="high")

# 添加历史对话（普通优先级）
manager.add("用户：...", priority="normal")
manager.add("助手：...", priority="normal")

# 添加临时信息（低优先级）
manager.add("临时计算结果：...", priority="low")

# 自动截断低优先级内容
current_context = manager.get_context()
                        

                    ╱─────────────╲
                   ╱   E2E 测试    ╲     10%
                  ╱  (端到端测试)   ╲    覆盖完整流程
                 ╱─────────────────╲
                ╱    集成测试       ╲   20%
               ╱  (组件交互测试)     ╲  测试组件协作
              ╱─────────────────────╲
             ╱      单元测试         ╲  70%
            ╱    (组件独立测试)       ╲ 测试单个组件
           ╱─────────────────────────╲
          
          测试数量：左多右少
          执行速度：左快右慢
          调试难度：左易右难
          置信度：   左低右高

          推荐比例：
            • 单元测试：70% - 快速、隔离、确定性
            • 集成测试：20% - 组件交互、接口验证
            • E2E 测试：10% - 完整流程、真实场景
                        

import pytest
from tools import SearchTool, CalculatorTool

class TestSearchTool:
    @pytest.fixture
    def search_tool(self):
        return SearchTool(api_key="test_key")
    
    def test_search_success(self, search_tool):
        """测试搜索成功"""
        result = search_tool.run("2026 AI 趋势")
        assert isinstance(result, str)
        assert len(result) > 0
        assert "2026" in result or "AI" in result
    
    def test_search_empty_query(self, search_tool):
        """测试空查询"""
        with pytest.raises(ValueError, match="查询不能为空"):
            search_tool.run("")
    
    def test_search_timeout(self, search_tool, mocker):
        """测试超时处理"""
        mocker.patch("requests.get", side_effect=TimeoutError)
        result = search_tool.run("测试", timeout=1)
        assert "超时" in result
    
    def test_search_rate_limit(self, search_tool, mocker):
        """测试速率限制"""
        mocker.patch("requests.get", side_effect=Exception("429 Too Many Requests"))
        result = search_tool.run("测试")
        assert "速率限制" in result

class TestCalculatorTool:
    @pytest.fixture
    def calc_tool(self):
        return CalculatorTool()
    
    def test_addition(self, calc_tool):
        assert calc_tool.run("2 + 2") == "4"
    
    def test_division_by_zero(self, calc_tool):
        result = calc_tool.run("1 / 0")
        assert "错误" in result
        assert "除零" in result
    
    def test_invalid_expression(self, calc_tool):
        result = calc_tool.run("abc + def")
        assert "错误" in result