C3-04 零框架理解 Agent：ReAct 手撕

def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        allowed = set("0123456789+-*/().% ")
        if all(c in allowed for c in expression):
            return str(eval(expression))
        return "不支持的表达式"
    except Exception as e:
        return f"计算错误: {e}"


def search(query: str) -> str:
    """模拟搜索引擎"""
    knowledge = {
        "python": "Python 是一门解释型高级编程语言,1991 年由 Guido 创建",
        "transformer": "Transformer 是 2017 年 Google 提出的注意力机制架构",
        "rag": "RAG(检索增强生成)是 2020 年 Facebook 提出的技术",
        "agent": "AI Agent 是能自主感知、规划、行动的智能体系统",
    }
    for key, value in knowledge.items():
        if key in query.lower():
            return value
    return f"未找到 '{query}' 的相关信息"


def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


# 工具注册表
TOOLS = {
    "calculator": {"func": calculator, "desc": "计算数学表达式,参数: expression"},
    "search": {"func": search, "desc": "搜索信息,参数: query"},
    "get_time": {"func": get_current_time, "desc": "获取当前时间,无参数"},
}

你是一个智能AI助手,可以使用以下工具来解决问题:

{tools_desc}

请严格按照以下格式思考和行动:

Thought: 我需要思考如何回答这个问题
Action: 工具名(参数)
Observation: [工具返回结果,由系统填入]
... (可以重复 Thought/Action/Observation 多次)
Thought: 我现在知道答案了
Final Answer: 最终回答

重要规则:
1. 每次只调用一个工具
2. Action 格式必须是: 工具名(参数)
3. 如果不需要工具,直接给出 Final Answer
4. 最多执行 5 轮工具调用

开始!

问题: {question}

import re
import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

REACT_PROMPT = """你是一个智能AI助手,可以使用以下工具来解决问题:

{tools_desc}

请严格按照以下格式思考和行动:

Thought: 我需要思考如何回答这个问题
Action: 工具名(参数)
Observation: [工具返回结果,由系统填入]
... (可以重复 Thought/Action/Observation 多次)
Thought: 我现在知道答案了
Final Answer: 最终回答

重要规则:
1. 每次只调用一个工具
2. Action 格式必须是: 工具名(参数)
3. 如果不需要工具,直接给出 Final Answer
4. 最多执行 5 轮工具调用

开始!

问题: {question}
"""


class ReActAgent:
    """手撕 ReAct Agent - 零框架实现"""

    def __init__(self, tools: dict, model: str = None, max_steps: int = 5):
        self.tools = tools
        self.model = model or os.getenv("MODEL_NAME", "gpt-4o-mini")
        self.max_steps = max_steps
        self.client = OpenAI(
            api_key=os.getenv("OPENAI_API_KEY"),
            base_url=os.getenv("OPENAI_BASE_URL"),
        )

    def _build_tools_desc(self) -> str:
        lines = [f"- {name}: {info['desc']}" for name, info in self.tools.items()]
        return "\n".join(lines)

    def _parse_action(self, text: str):
        """从 LLM 输出中解析 Action(正则匹配)"""
        match = re.search(r"Action:\s*(\w+)\((.*?)\)", text)
        if match:
            return match.group(1), match.group(2).strip().strip('"').strip("'")
        return None, None

    def _check_final_answer(self, text: str):
        match = re.search(r"Final Answer:\s*(.+)", text, re.DOTALL)
        return match.group(1).strip() if match else None

    def _call_llm(self, prompt: str) -> str:
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
            stop=["Observation:"],  # 关键:让 LLM 在需要观察结果时暂停
        )
        return response.choices[0].message.content

    def run(self, question: str) -> str:
        prompt = REACT_PROMPT.format(
            tools_desc=self._build_tools_desc(),
            question=question,
        )
        conversation = prompt

        for step in range(self.max_steps):
            llm_output = self._call_llm(conversation)
            print(f"\n--- 步骤 {step + 1} ---")
            print(llm_output.strip())

            final = self._check_final_answer(llm_output)
            if final:
                return final

            tool_name, tool_arg = self._parse_action(llm_output)
            if tool_name and tool_name in self.tools:
                result = (
                    self.tools[tool_name]["func"](tool_arg)
                    if tool_arg
                    else self.tools[tool_name]["func"]()
                )
                print(f"Observation: {result}")
                conversation += llm_output + f"\nObservation: {result}\n"
            else:
                conversation += (
                    llm_output + "\nObservation: 工具调用失败,请按格式重新输出\n"
                )

        return "达到最大步骤数,未能得出答案"


if __name__ == "__main__":
    from agent_course_demo import TOOLS  # 实际使用时直接 import 上一节的工具
    agent = ReActAgent(tools=TOOLS)

    questions = [
        "计算 (15 + 27) * 3 的结果",
        "什么是 AI Agent?",
        "北京有多少人口?用计算器算出如果每人每天消耗 2 升水,北京一天需要多少升水?",
    ]

    for q in questions:
        try:
            agent.run(q)
        except Exception as e:
            print(f"错误: {e}")
            break

--- 步骤 1 ---
Thought: 用户要求计算 (15+27)*3 的结果,这是个数学表达式。
Action: calculator((15 + 27) * 3)
Observation: 126
--- 步骤 2 ---
Thought: 我已经计算出结果 126,可以给出最终答案了。
Final Answer: (15 + 27) * 3 的结果是 126。