Agent Loop关键日揭秘：Day5为何是成败分水岭

2026-06-09阅读 0热度 0

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）。两种模式各有优劣，适配不同场景。