今日目标

把前几天的零散知识整合成一个生产级可用的 Agent，具备：

• 完整的工具调用循环
• 错误处理和重试机制
• 步数限制防止无限循环
• 清晰的执行日志

---

核心概念：ReAct 模式

ReAct = Reasoning + Acting
用户问题
↓
┌─────────────────────────┐
│ Thought: 我需要先查天气 │ ← 推理
│ Action: get_weather │ ← 行动
│ Action Input: 北京 │
└─────────────────────────┘
↓
执行工具，得到结果
↓
┌─────────────────────────┐
│ Observation: 晴，25°C │ ← 观察
│ Thought: 我已经知道答案 │
│ Final Answer: 北京晴天 │ ← 最终回答
└─────────────────────────┘

---

完整代码实现

第一部分：工具注册系统

from dataclasses import dataclass
from typing import Callable, Dict, Any, Optional
import json

@dataclass
class Tool:
    """工具定义"""
    name: str
    description: str
    func: Callable
    parameters: Dict[str, str]  # {参数名: 参数说明}

class ToolRegistry:
    """工具注册中心"""
    
    def __init__(self):
        self.tools: Dict[str, Tool] = {}
    
    def register(self, name: str, description: str, parameters: Optional[Dict[str, str]] = None):
        """装饰器方式注册工具"""
        def decorator(func: Callable):
            self.tools[name] = Tool(
                name=name,
                description=description,
                func=func,
                parameters=parameters or {}
            )
            return func
        return decorator
    
    def get_tool(self, name: str) -> Optional[Tool]:
        return self.tools.get(name)
    
    def execute(self, name: str, **kwargs) -> str:
        """执行工具，带错误处理"""
        tool = self.get_tool(name)
        if not tool:
            return f"错误：工具 '{name}' 不存在"
        
        try:
            result = tool.func(**kwargs)
            return str(result)
        except TypeError as e:
            return f"参数错误：{e}"
        except Exception as e:
            return f"执行错误：{e}"
    
    def get_tools_prompt(self) -> str:
        """生成工具描述，供 LLM 使用"""
        if not self.tools:
            return "当前没有可用工具。"
        
        lines = ["你可以使用以下工具：", ""]
        for tool in self.tools.values():
            lines.append(f"**{tool.name}**: {tool.description}")
            if tool.parameters:
                params = ", ".join([f"{k}({v})" for k, v in tool.parameters.items()])
                lines.append(f"  参数: {params}")
            lines.append("")
        return "\n".join(lines)


# 创建全局注册器
registry = ToolRegistry()

第二部分：注册工具

import datetime

@registry.register(
    name="get_time",
    description="获取当前时间",
    parameters={}
)
def get_time() -> str:
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")


@registry.register(
    name="get_weather",
    description="获取指定城市的天气",
    parameters={"city": "城市名称"}
)
def get_weather(city: str) -> str:
    # 模拟天气数据
    weather_data = {
        "北京": "晴，气温 18-26°C，空气质量良",
        "上海": "多云，气温 20-28°C，空气质量优",
        "广州": "小雨，气温 22-30°C，空气质量良",
    }
    return weather_data.get(city, f"未找到 {city} 的天气信息")


@registry.register(
    name="calculator",
    description="计算数学表达式",
    parameters={"expression": "数学表达式，如 2+3*4"}
)
def calculator(expression: str) -> str:
    try:
        # 安全限制：只允许数字和基本运算符
        allowed = set("0123456789+-*/().% ")
        if not all(c in allowed for c in expression):
            return "错误：表达式包含非法字符"
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误：{e}"


@registry.register(
    name="search",
    description="搜索信息（模拟）",
    parameters={"query": "搜索关键词"}
)
def search(query: str) -> str:
    # 模拟搜索结果
    mock_results = {
        "python": "Python 是一种流行的编程语言，由 Guido van Rossum 创建",
        "ai": "人工智能（AI）是计算机科学的一个分支，致力于创建智能机器",
        "langchain": "LangChain 是一个用于构建 LLM 应用的框架",
    }
    query_lower = query.lower()
    for key, value in mock_results.items():
        if key in query_lower:
            return value
    return f"未找到与 '{query}' 相关的结果"

第三部分：ReAct Agent 核心

from openai import OpenAI
import re

class ReActAgent:
    """ReAct 模式的 Agent 实现"""
    
    def __init__(self, registry: ToolRegistry, api_key: str, model: str = "gpt-4o-mini"):
        self.registry = registry
        self.client = OpenAI(api_key=api_key)
        self.model = model
        self.max_steps = 10  # 最大循环次数
    
    def _build_system_prompt(self) -> str:
        """构建系统提示词"""
        tools_desc = self.registry.get_tools_prompt()
        
        return f"""你是一个有用的 AI 助手，可以使用工具来帮助回答问题。

{tools_desc}

请严格按照以下格式回复：

如果需要使用工具：
Thought: [你的思考过程]
Action: [工具名称]
Action Input: [工具参数，JSON 格式]

如果已经得到最终答案：
Thought: [你的思考过程]
Final Answer: [最终回答]

重要规则：
1. 每次只能使用一个工具
2. Action Input 必须是合法的 JSON 格式
3. 如果工具返回错误，尝试修正或换一种方法
4. 不要编造工具不存在的信息"""
    
    def _parse_response(self, text: str) -> Dict[str, Any]:
        """解析 LLM 的回复"""
        result = {
            "thought": None,
            "action": None,
            "action_input": None,
            "final_answer": None
        }
        
        # 提取 Thought
        thought_match = re.search(r"Thought:\s*(.+?)(?=\n(?:Action|Final Answer)|$)", text, re.DOTALL)
        if thought_match:
            result["thought"] = thought_match.group(1).strip()
        
        # 提取 Final Answer
        final_match = re.search(r"Final Answer:\s*(.+?)$", text, re.DOTALL)
        if final_match:
            result["final_answer"] = final_match.group(1).strip()
            return result
        
        # 提取 Action 和 Action Input
        action_match = re.search(r"Action:\s*(.+?)(?=\n|$)", text)
        if action_match:
            result["action"] = action_match.group(1).strip()
        
        input_match = re.search(r"Action Input:\s*(.+?)(?=\n(?:Thought|Action|$)|$)", text, re.DOTALL)
        if input_match:
            raw_input = input_match.group(1).strip()
            try:
                result["action_input"] = json.loads(raw_input)
            except json.JSONDecodeError:
                # 尝试简单参数（非 JSON）
                result["action_input"] = {"value": raw_input}
        
        return result
    
    def _call_llm(self, messages: list) -> str:
        """调用 LLM"""
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0
        )
        return response.choices[0].message.content
    
    def run(self, user_input: str, verbose: bool = True) -> str:
        """执行 Agent 循环"""
        
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            {"role": "user", "content": user_input}
        ]
        
        if verbose:
            print(f"\n{'='*50}")
            print(f"用户问题: {user_input}")
            print('='*50)
        
        for step in range(self.max_steps):
            if verbose:
                print(f"\n--- Step {step + 1} ---")
            
            # 调用 LLM
            response_text = self._call_llm(messages)
            
            if verbose:
                print(f"LLM 回复:\n{response_text}")
            
            # 解析回复
            parsed = self._parse_response(response_text)
            
            # 检查是否有最终答案
            if parsed["final_answer"]:
                if verbose:
                    print(f"\n✅ 最终答案: {parsed['final_answer']}")
                return parsed["final_answer"]
            
            # 检查是否有工具调用
            if parsed["action"]:
                action = parsed["action"]
                action_input = parsed["action_input"] or {}
                
                if verbose:
                    print(f"🔧 执行工具: {action}")
                    print(f"   参数: {action_input}")
                
                # 执行工具
                observation = self.registry.execute(action, **action_input)
                
                if verbose:
                    print(f"📋 执行结果: {observation}")
                
                # 将结果添加到对话历史
                messages.append({"role": "assistant", "content": response_text})
                messages.append({"role": "user", "content": f"Observation: {observation}"})
            
            else:
                # 既没有 Final Answer 也没有 Action，异常情况
                if verbose:
                    print("⚠️ 解析失败，要求模型重新回复")
                messages.append({"role": "assistant", "content": response_text})
                messages.append({"role": "user", "content": "请按照指定格式回复，使用 Action 或 Final Answer。"})
        
        # 超过最大步数
        if verbose:
            print(f"\n❌ 达到最大步数 ({self.max_steps})，强制结束")
        return "抱歉，我无法在有限步骤内完成这个任务。"

第四部分：测试运行

# 创建 Agent
agent = ReActAgent(
    registry=registry,
    api_key="你的OpenAI API Key",
    model="gpt-4o-mini"
)

# 测试单工具调用
print("\n" + "="*60)
print("测试 1: 简单查询")
result = agent.run("现在几点了？")

# 测试多工具调用
print("\n" + "="*60)
print("测试 2: 多步骤任务")
result = agent.run("北京和上海的天气分别怎么样？哪个城市更热？")

# 测试计算
print("\n" + "="*60)
print("测试 3: 计算任务")
result = agent.run("计算 (15 + 27) * 3 等于多少")

# 测试搜索
print("\n" + "="*60)
print("测试 4: 搜索任务")
result = agent.run("什么是 LangChain？")

---

扩展练习

练习 1：添加重试机制

工具执行失败时自动重试：

def execute_with_retry(self, name: str, max_retries: int = 3, **kwargs) -> str:
    for attempt in range(max_retries):
        result = self.execute(name, **kwargs)
        if not result.startswith("错误") and not result.startswith("参数错误"):
            return result
        if attempt < max_retries - 1:
            print(f"重试 {attempt + 2}/{max_retries}...")
    return result

练习 2：添加执行历史记录

@dataclass
class ExecutionStep:
    step_num: int
    thought: str
    action: Optional[str]
    action_input: Optional[Dict]
    observation: Optional[str]
    timestamp: datetime.datetime

class ReActAgent:
    def __init__(self, ...):
        ...
        self.history: list[ExecutionStep] = []
    
    def get_history(self) -> list[ExecutionStep]:
        return self.history

练习 3：支持 Claude API

如果你用的是 Anthropic API：

from anthropic import Anthropic

class ClaudeReActAgent(ReActAgent):
    def __init__(self, registry: ToolRegistry, api_key: str):
        self.registry = registry
        self.client = Anthropic(api_key=api_key)
        self.model = "claude-sonnet-4-20250514"
        self.max_steps = 10
    
    def _call_llm(self, messages: list) -> str:
        # 分离 system 和其他消息
        system_msg = ""
        other_msgs = []
        for msg in messages:
            if msg["role"] == "system":
                system_msg = msg["content"]
            else:
                other_msgs.append(msg)
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=system_msg,
            messages=other_msgs
        )
        return response.content[0].text

今日检查清单

• 理解 ReAct 的 Thought → Action → Observation 循环
• 能跑通完整的 Agent 代码
• 理解 max_steps 为什么重要（防止无限循环）
• 理解错误处理在哪些环节发生
• 尝试添加一个自己的工具

---

明日预告
Day 15：LangChain 入门
用框架重写今天的 Agent，你会发现：

• Tool → LangChain 的 @tool 装饰器
• ReActAgent.run() → LangChain 的 AgentExecutor
• 手动解析 → LangChain 的 OutputParser
  有了今天的基础，学框架会快很多。

---

有问题随时问！