7天从零手搓 AI Agent | Day 5:Agent Loop——最关键的一天

Agent 和普通程序的根本区别:Agent 能自己决定下一步做什么,而不是等人告诉它。


今天的目标

理解并实现 Agent Loop——Agent 系统中最核心的机制。

之前我们写的代码,本质上是"人驱动"的:

用户输入 → AI 决策 → 执行 → 结束

今天,我们要写"自驱动"的 Agent Loop:

用户输入 → AI 思考 → 执行 → AI 再思考 → 执行 → ... → 任务完成

AI 自己决定什么时候停。


为什么需要 Loop

考虑这个任务:“帮我检查当前目录有哪些 Python 文件,统计每个文件有多少行代码。”

分步来看:

  1. 列出当前目录的文件
  2. 筛选出 .py 文件
  3. 逐个读取每个文件
  4. 统计每个文件的行数
  5. 汇总结果

如果用 Day 3 的方式(AI 预先规划所有步骤),AI 必须在执行前就知道一共有多少个 Python 文件,才能规划出正确的步骤数。

但 AI 不知道。它需要先看到结果,再决定下一步

这就是 Agent Loop 解决的问题。


Agent Loop 的核心思想

┌──────────────────────────────────────┐
│                                      │
│   ┌─────────┐                        │
│   │  观察    │ ← 环境状态 + 工具结果  │
│   └────┬────┘                        │
│        ↓                             │
│   ┌─────────┐                        │
│   │  思考    │ ← AI 分析当前情况      │
│   └────┬────┘                        │
│        ↓                             │
│   ┌─────────┐                        │
│   │  行动    │ ← 调用工具 / 给出回答  │
│   └────┬────┘                        │
│        ↓                             │
│   ┌─────────┐                        │
│   │  停止?  │ → 是 → 结束            │
│   └────┬────┘                        │
│        │ 否                          │
│        └──→ 回到"观察"               │
│                                      │
└──────────────────────────────────────┘

关键:每一步的输出,都会反馈给 AI 作为下一步的输入。


完整代码

工具实现参考 Day 2。这里只展示 Agent Loop 核心逻辑。

# day5_agent_loop.py

import json
import inspect
from openai import OpenAI

client = OpenAI(
    api_key="你的API Key",
    base_url="https://api.mimo.ai/v1",
)


def think(user_input: str, messages: list) -> dict:
    """AI 思考并决策"""
    tools_prompt = build_tools_prompt()  # 从 Day 2 导入

    system_prompt = f"""你是一个智能助手,可以使用工具来完成任务。

{tools_prompt}

工作流程:
1. 分析用户需求
2. 如果需要工具,调用工具获取信息
3. 根据工具结果,决定下一步
4. 如果任务完成,给出最终回答

重要:每一步只调用一个工具。根据工具返回的结果,再决定下一步。

JSON 格式:
- 调用工具:{{"thought": "思考", "action": "tool", "tool": "工具名", "params": {{}}}}
- 最终回答:{{"thought": "思考", "action": "final", "response": "回答"}}

只输出 JSON。"""

    full_messages = [{"role": "system", "content": system_prompt}]
    full_messages.extend(messages)

    response = client.chat.completions.create(
        model="mimo-v2-flash",
        messages=full_messages,
        temperature=0,
    )

    reply = response.choices[0].message.content.strip()
    try:
        if reply.startswith("```"):
            reply = reply.split("\n", 1)[1]
            reply = reply.rsplit("```", 1)[0]
        return json.loads(reply)
    except json.JSONDecodeError:
        return {"thought": "解析失败", "action": "final", "response": reply}


def agent_loop(user_input: str, max_iterations: int = 10):
    """
    Agent 的核心循环

    这就是 Agent 和普通程序的根本区别:
    - 普通程序:输入 → 处理 → 输出
    - Agent Loop:输入 → 思考 → 行动 → 观察 → 思考 → 行动 → ... → 输出
    """
    print(f"\n{'='*50}")
    print(f"你:{user_input}")
    print(f"{'='*50}")

    messages = [{"role": "user", "content": user_input}]

    for iteration in range(1, max_iterations + 1):
        print(f"\n[步骤 {iteration}/{max_iterations}]")

        # 思考
        decision = think(user_input, messages)
        thought = decision.get("thought", "")
        action = decision.get("action", "final")

        print(f"[AI 思考]: {thought}")

        # 如果是最终回答,结束循环
        if action == "final":
            response = decision.get("response", "无法生成回答。")
            print(f"\nAgent:{response}")
            print(f"[完成,共 {iteration} 步]")
            return response

        # 如果是工具调用
        if action == "tool":
            tool_name = decision.get("tool", "")
            params = decision.get("params", {})

            print(f"[调用工具]: {tool_name}")

            # 执行工具
            result = execute_tool(tool_name, params)  # 从 Day 2 导入
            print(f"[工具结果]: {result[:200]}{'...' if len(result) > 200 else ''}")

            # 把工具调用和结果加入消息历史
            # 这样 AI 下次思考时能看到之前做了什么
            messages.append({
                "role": "assistant",
                "content": json.dumps(decision, ensure_ascii=False),
            })
            messages.append({
                "role": "user",
                "content": f"工具 {tool_name} 的执行结果:\n{result}",
            })
        else:
            print(f"[未知动作: {action},强制结束]")
            break

    # 达到最大迭代次数
    print(f"\n[达到最大迭代次数 {max_iterations},强制结束]")
    print("Agent:抱歉,任务比较复杂,我需要更多步骤才能完成。请尝试简化你的需求。")
    return None

运行测试:真正的自主决策

你:帮我检查当前目录有哪些 Python 文件,统计每个文件有多少行代码

==================================================
你:帮我检查当前目录有哪些 Python 文件,统计每个文件有多少行代码
==================================================

[步骤 1/10]
[AI 思考]: 用户想知道当前目录的 Python 文件及其行数。首先需要列出文件。
[调用工具]: list_files
[工具结果]: 📄 day1_basic.py (1234 bytes)
📄 day2_tools.py (2345 bytes)
📄 day3_multi_tools.py (3456 bytes)

[步骤 2/10]
[AI 思考]: 找到了3个Python文件。现在需要逐个统计行数。先统计第一个。
[调用工具]: count_lines
[工具结果]: 文件 day1_basic.py:共 50 行,其中代码 45 行,空行 5 行

[步骤 3/10]
[AI 思考]: 第一个文件统计完成。继续统计第二个。
[调用工具]: count_lines
[工具结果]: 文件 day2_tools.py:共 80 行,其中代码 72 行,空行 8 行

[步骤 4/10]
[AI 思考]: 第二个文件统计完成。继续统计第三个。
[调用工具]: count_lines
[工具结果]: 文件 day3_multi_tools.py:共 110 行,其中代码 98 行,空行 12 行

[步骤 5/10]
[AI 思考]: 所有文件都统计完成了,现在可以给用户总结了。

Agent:当前目录共有 3 个 Python 文件,统计如下:

| 文件名 | 总行数 | 代码行数 | 空行数 |
|--------|--------|----------|--------|
| day1_basic.py | 50 | 45 | 5 |
| day2_tools.py | 80 | 72 | 8 |
| day3_multi_tools.py | 110 | 98 | 12 |
| **合计** | **240** | **215** | **25** |

[完成,共 5 步]

核心概念:Loop 为什么强大

对比:Day 3 vs Day 5

Day 3(预规划) Day 5(Agent Loop)
步骤 AI 在执行前规划所有步骤 AI 每一步都重新评估
灵活性 固定的步骤列表 动态调整
错误处理 步骤失败就失败 AI 可以换方案
适用场景 步骤数固定、可预测 复杂、不确定的任务

Agent Loop 的三个关键点

1. 观察是动态的

messages.append({
    "role": "user",
    "content": f"工具 {tool_name} 的执行结果:\n{result}",
})

AI 看到的不只是用户最初的输入,还有每一步的工具结果。这就像一个人做事情:先看第一步的结果,再决定第二步怎么做。

2. 停止条件是 AI 决定的

if action == "final":
    response = decision.get("response", "无法生成回答。")
    print(f"\nAgent:{response}")
    return response

AI 自己判断任务是否完成。它不需要被明确告知"你做了5步了,该停了"。

3. 有兜底机制

for iteration in range(1, max_iterations + 1):
    # ...
    # 如果 AI 一直不停,达到最大迭代次数强制结束

max_iterations 是安全阀。没有它,AI 可能陷入死循环。


一个真实的错误恢复案例

你:帮我读取 test.txt 文件的内容

[步骤 1/10]
[AI 思考]: 用户想读取 test.txt 文件
[调用工具]: read_file
[工具结果]: 错误:文件 'test.txt' 不存在

[步骤 2/10]
[AI 思考]: 文件不存在。我应该先看看当前目录有什么文件,然后告诉用户。
[调用工具]: list_files
[工具结果]: 📄 notes.txt (100 bytes)
📄 readme.md (200 bytes)

[步骤 3/10]
[AI 思考]: 当前目录没有 test.txt,但有 notes.txt 和 readme.md。告诉用户。

Agent:当前目录中没有 test.txt 文件。现有的文件有:
- notes.txt (100 bytes)
- readme.md (200 bytes)

你是不是想读取其中一个?
[完成,共 3 步]

Agent 没有在工具失败时崩溃,而是自己找到了替代方案。

这就是 Agent Loop 的威力:失败不是终点,而是新的输入。


常见错误

错误 1:忘记把工具结果加入消息历史

# 错误:AI 看不到之前做了什么
result = execute_tool(tool_name, params)
# 就这样结束了,没有把 result 加入 messages

# 正确:把结果加入消息历史
messages.append({
    "role": "user",
    "content": f"工具 {tool_name} 的执行结果:\n{result}",
})

错误 2:max_iterations 太小

# 错误:复杂任务还没完成就被强制结束了
max_iterations = 3  # 对于需要 5 步的任务来说太小

# 正确:根据任务复杂度设置
max_iterations = 10  # 给 AI 足够的空间

错误 3:没有处理 AI 返回的未知动作

# 错误:如果 AI 返回 "action": "unknown",程序会卡住
if action == "tool":
    # ...
elif action == "final":
    # ...
# 没有 else 分支!

# 正确:处理所有情况
else:
    print(f"[未知动作: {action},强制结束]")
    break

动手实验

  1. max_iterations 改成 3,观察 Agent 在复杂任务中的行为
  2. 故意让一个工具总是失败,看 Agent 怎么恢复
  3. 加一个新工具,看 Agent 能不能自己学会使用

今天的洞察

Agent 的本质不是"调用 AI",而是"让 AI 驱动一个循环"。

AI 本身只是一个文本生成器。Agent Loop 赋予了它"做事"的能力。


明天预告

今天我们有了能自主执行的 Agent,但它还是"一步一动"的。如果用户说"帮我写一份市场调研报告",Agent 会一个一个搜索,最后拼凑起来。

更好的方式是:先规划,再执行。就像你做项目一样,先列出大纲,再逐项填充。

Day 6:先规划,再执行

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐