Day 5：Agent Loop——整个系列里最关键的一天

Day 5：Agent Loop——整个系列里最关键的一天

今天是这个系列里技术含量最高的一天，但好消息是，它也是最值得搞清楚的一天。因为所有你听说过的Agent系统——AutoGPT、BabyAGI、Devin、Manus——它们的核心，都是今天要讲的这个循环。

前4天的问题

前4天，Agent的工作流程是这样的：

用户输入 → AI 决策（用哪个工具？） → 执行一次 → 输出结果

最多执行一次动作，就结束了。但真实任务往往需要多步：搜索新闻 → 看搜索结果 → 可能再补充搜索 → 整理和总结 → 分析趋势。一步做不完，需要循环。

ReAct：思考、行动、观察，反复循环

2022年，一篇名为ReAct（发音像英文单词react）的论文提出了一个思路：

Thought（思考）→ Action（行动）→ Observation（观察结果）→ Thought → ...

让AI反复执行这个三步循环，直到任务完成。这个思路很直观——人做复杂任务的时候也是这样：先搜一下（行动），哦有了一些信息（观察），但还不够，再搜一个关键词（行动），好了信息足够了，现在来总结（最终行动）。Agent Loop就是把人类的思考过程，写成代码。

实现ReAct Loop

新建agent_loop.py：

# agent_loop.py
import json
import re
from llm import chat
from tool_registry import get_tools_description, execute_tool
from memory.short_term import ShortTermMemory

REACT_SYSTEM_PROMPT = """你是一个能完成复杂任务的智能助手，可以反复使用工具直到任务完成。
{tools_description}

每次回复必须是 JSON，三种格式之一：
1. 需要使用工具（可以多次使用）：{{"type": "tool_call", "tool": "工具名", "params": {{"参数名": "参数值"}}, "thought": "我为什么用这个工具"}}
2. 任务已完成：{{"type": "final_answer", "content": "最终答案内容"}}
3. 需要向用户提问：{{"type": "ask_user", "question": "你的问题"}}

规则：
- 最多使用工具 {max_steps} 次
- 收集到足够信息后，必须给出 final_answer
- 不要用相同参数重复调用同一个工具
- 只返回 JSON
"""

def safe_parse_json(text: str) -> dict:
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    match = re.search(r'{.*}', text, re.DOTALL)
    if match:
        try:
            return json.loads(match.group())
        except json.JSONDecodeError:
            pass
    return {"type": "final_answer", "content": text}

class ReactAgent:
    def __init__(self, max_steps: int = 5) -> None:
        self.max_steps = max_steps
    
    def run(self, user_task: str) -> str:
        """运行 ReAct 循环完成任务，返回最终答案。"""
        # 每次任务新建一个记忆实例（不跨任务保留中间步骤）
        memory = ShortTermMemory(max_messages=40)
        memory.add("system", REACT_SYSTEM_PROMPT.format(
            tools_description=get_tools_description(),
            max_steps=self.max_steps,
        ))
        memory.add("user", f"请帮我完成这个任务：{user_task}")
        
        for step in range(1, self.max_steps + 1):
            print(f"n{'─'*40}")
            print(f"[步骤 {step}/{self.max_steps}]")
            
            ai_response = chat(memory.to_api_format())
            print(f"[AI 思考]: {ai_response}")
            memory.add("assistant", ai_response)
            
            decision = safe_parse_json(ai_response)
            resp_type = decision.get("type")
            
            # 任务完成
            if resp_type == "final_answer":
                print(f"[任务完成，共 {step} 步]")
                return decision.get("content", "（无内容）")
            
            # 使用工具
            if resp_type == "tool_call":
                tool_name = decision.get("tool", "")
                params= decision.get("params", {})
                thought = decision.get("thought", "")
                print(f"[调用工具]: {tool_name}")
                if thought:
                    print(f"[理由]: {thought}")
                result = execute_tool(tool_name, params)
                # 只打印前300字，避免终端被刷屏
                preview = result[:300] + "..." if len(result) > 300 else result
                print(f"[工具结果]: {preview}")
                # 把工具结果作为"观察"加入记忆
                memory.add("user", f"工具 {tool_name} 返回结果：n{result}")
                continue
            
            # 向用户提问
            if resp_type == "ask_user":
                question = decision.get("question", "")
                user_answer = input(f"nAgent 问你：{question}n你：")
                memory.add("user", user_answer)
                continue
            
            # 未知类型，结束
            return str(decision)
        
        # 达到步骤上限，强制要求给出答案
        print(f"n[已达步骤上限 {self.max_steps}，要求给出最终答案]")
        memory.add("user", "你已用完所有步骤，请立即基于已有信息给出最终答案。")
        final_response = chat(memory.to_api_format())
        final = safe_parse_json(final_response)
        return final.get("content", final_response)

核心逻辑只有这几行

整个ReAct的精髓，其实就是一个while循环里的三个分支：

if resp_type == "final_answer":
    return ...  # 任务完成，退出循环
if resp_type == "tool_call":
    result = 执行工具
    memory.add(result)  # 把结果加入记忆，下一步 AI 能看到
    continue  # 继续循环
if resp_type == "ask_user":
    answer = input(...)
    memory.add(answer)
    continue

每一步，AI都能看到之前所有步骤的结果（因为都存在memory里），所以它能做出更好的决策。

运行效果

你：帮我搜索最近 AI 领域的重要新闻，总结3条最重要的

────────────────────────────────────────
[步骤 1/5]
[AI 思考]: {"type": "tool_call", "tool": "web_search", "params": {"query": "AI 人工智能最新新闻 2024"}, "thought": "需要先搜索最新 AI 新闻"}
[调用工具]: web_search
[理由]: 需要先搜索最新 AI 新闻
[工具结果]: 摘要：人工智能领域...
────────────────────────────────────────
[步骤 2/5]
[AI 思考]: {"type": "final_answer", "content": "根据搜索结果，以下是3条最重要的 AI 新闻：nn1. ..."}
[任务完成，共 2 步]
Agent：根据搜索结果，以下是3条最重要的 AI 新闻：
1. ...
2. ...
3. ...

max_steps有多重要

假设没有步骤上限，会发生什么？有时候AI会陷入一种状态：觉得自己信息不够，一直在搜索，搜索，再搜索……永远不输出答案。这叫死循环。

max_steps=5是一道保险：最多执行5步，然后强制要求给出答案。即使AI觉得信息不够，也要用现有的信息给一个答案。一般任务用5步就够了，复杂的研究任务可以设8-10步。记住，每步都要调用API，步骤越多，费用越高。

thought字段有什么用

注意tool_call类型的JSON里有一个thought字段：

{"type": "tool_call", "tool": "web_search", "params": {...}, "thought": "我为什么要用这个工具"}

这是让AI说出自己的思考过程。好处有两点：调试方便——你能看到AI为什么做这个决定，更容易发现问题；决策更准确——先说出理由再行动，AI的准确率会提高（这是Chain-of-Thought的效果）。如果Agent做了奇怪的决定，先看thought，通常能找到原因。

今天的项目结构

my_agent/
├── .env
├── llm.py
├── agent_loop.py   # 新增：ReAct 循环（今天最重要的文件）
├── tool_registry.py
├── memory/
│   └── short_term.py
├── tools/
│   ├── search.py
│   ├── weather.py
│   ├── calculator.py
│   └── datetime_tool.py
└── main.py

小结

今天的ReAct循环，是整个系列里最值得理解透的部分。核心只有一句话：把工具结果存入记忆，让AI在下一步能看到，然后继续决策——直到任务完成或达到步骤上限。

这个模式，你在LangChain里看到的AgentExecutor，在OpenAI的Assistants API里看到的Run Loop，都是它的变体。

明天，Day 6：先想清楚再动手——给Agent加上规划能力。ReAct是"边做边想"，明天我们学"先想好再做"（Plan-and-Execute）。两种模式各有优势，适合不同场景。