📌 前置知识:已完成第一课至第五课
🎯 本课目标:把决策和工具调用放进循环,让 AI 反复思考、反复行动
💡 核心概念:Agent Loop / 状态追踪 / 终止条件 / 步骤累积


前言

前五课,我们一步步给 AI 加了能力——

第二课给了它角色,第三课给了它结构化输出,第四课给了它决策能力,第五课给了它工具调用。

但不管是哪一课,AI 的行为模式都一样:

用户说话 → AI 回复一次 → 结束

一回。就这一回。

ChatGPT 是这么工作的吗?不是。当你让 ChatGPT “帮我查一下明天北京的天气,然后推荐适合的穿搭”,它不会一步到位——它先查天气,看到"晴,25℃",然后根据这个结果推荐穿搭。

中间发生了一次观察→决策→执行的循环。它把上一步的结果当作下一步的输入,一步步推进,直到任务完成。

这就是 Agent 和 Chatbot 的核心区别:Agent 不止回答一次,它会循环。


一、从"问答"到"循环"

先看一下我们前五课构建的系统是什么样子:

# 第四课
decision = agent.decide("帮我总结这篇文章", choices=["answer", "summarize", "translate"])
# AI 选了 "summarize",然后?没了。

# 第五课
tool_call = agent.request_tool("What is 42 * 7?")
result = agent.execute_tool_call(tool_call)
# AI 用了计算器,得到了 294,然后?也没了。

每次 AI 只做一件事,做完就结束。它的世界是"单步"的——看不到前一步的结果,也想不到下一步该做什么。

但真实的任务往往是多步的:

用户:帮我分析这篇文章的情感倾向,如果是负面的,总结主要原因

正确的处理流程:
  步骤1:分析情感 → "负面"
  步骤2:判断 → 是负面的,需要继续
  步骤3:总结原因 → "主要原因有3点……"
  步骤4:任务完成

而前五课的系统能做到吗?不能。因为它没有"上一步的结果"这个概念。

要支持多步任务,我们需要两样东西:

  1. 循环——让 AI 能反复思考、反复行动
  2. 状态——让 AI 知道"当前做到哪一步了,上一步做了什么"

第六课就做这两件事。


二、智能体循环:观察、决策、执行、重复

2.1 什么是 Agent Loop?

Agent Loop 就是把"观察→决策→执行"这个过程放在一个循环里,让 AI 重复执行,直到任务完成。

┌──────────────────────────────────┐
│            Agent Loop            │
│                                  │
│   ┌─────────────────────┐       │
│   │  1. 观察当前状态      │       │
│   │     (步骤计数、      │       │
│   │      前几步的结果)   │       │
│   └──────────┬──────────┘       │
│              ↓                   │
│   ┌─────────────────────┐       │
│   │  2. 决策下一步做什么   │       │
│   │     (调用 LLM)      │       │
│   └──────────┬──────────┘       │
│              ↓                   │
│   ┌─────────────────────┐       │
│   │  3. 执行动作          │       │
│   │     (更新状态)      │       │
│   └──────────┬──────────┘       │
│              ↓                   │
│        是否完成?                 │
│        ↓         ↓               │
│       否 → 回到 1    是 → 结束   │
│                                  │
└──────────────────────────────────┘

就这么简单。没有魔法,没有复杂框架。就是一个 while 循环,里面放三步逻辑。

你可能觉得"这也太简单了吧?"——确实简单。但简单不代表没用。这个循环模式几乎是所有 Agent 框架的底层骨架。LangChain 的 AgentExecutor、AutoGen 的对话循环、CrewAI 的任务编排,底层都是这个东西。

2.2 和前几课的对比

把五课放在一起对比,你能看到一条清晰的演进线:

第二课:用户 → AI 回复 → 结束
第三课:用户 → AI 返回 JSON → 结束
第四课:用户 → AI 选择动作 → 执行一次 → 结束
第五课:用户 → AI 选择工具+参数 → 执行一次 → 结束
第六课:用户 → AI 选择动作 → 执行 → 观察结果 → 再选择 → 再执行 → … → 完成

前五课是单步的,第六课是多步的。区别就在"循环"两个字。


三、状态:让循环有记忆

3.1 为什么需要状态?

没有状态的循环是瞎循环

想象一下:AI 在步骤 2 说"我要分析情感",在步骤 3 又说"我要分析情感",在步骤 4 还是"分析情感"——每一步都在做同一件事,因为它不记得上一步做过什么。

状态解决了这个问题。每次循环迭代,AI 都能看到当前的状态信息:

  • 已经执行了几步
  • 前面几步做了什么
  • 任务是否完成

有了这些信息,AI 才能做出合理的下一步决策。

3.2 状态包含什么?

最基本的状态只需要两个字段:

class AgentState:
    def __init__(self):
        self.steps = 0      # 已执行步数
        self.done = False    # 是否完成
        self.results = []    # 每步的结果(累积)
  • steps:防止无限循环。不管 AI 怎么想,步数到了就必须停。
  • done:AI 自己判断任务完成了,主动退出。
  • results:保存每步的产出,让后续步骤能看到前面的结果。

这比很多人想象的简单得多。不需要复杂的记忆系统(那是第七课的事),只需要一个计数器、一个标志位、一个列表。

3.3 状态放在哪?

放在 Prompt 里。

每次调用 LLM 之前,把当前状态拼进 prompt,让模型知道"你现在在第几步,前面做了什么"。

prompt = f"""...(角色设定)

当前状态:步骤={state.steps},已完成={state.done}
历史动作:{state.results}

请决定下一步动作..."""

就是这么直白。没有什么花哨的状态编码,没有什么向量数据库。直接把状态文字拼进 prompt,模型就能理解。


四、终止条件:没有刹车的车不能开

循环最怕什么?停不下来。

没有终止条件的循环会无限运行,把你的 API 配额烧光,把你的服务器跑死。所以终止条件是 Agent Loop 里最重要的安全机制。

本课实现三个终止条件,构成双重保险

4.1 AI 主动终止

AI 在某一步决定"任务完成了",输出 done 动作。

if action.get("action") == "done":
    state.mark_done()  # 标记完成,循环退出

这是理想情况——AI 认为任务做完了,主动退出。

4.2 最大步数限制

while not state.done and state.steps < max_steps:
    ...

不管 AI 怎么想,步数到了就强制停。这是你的安全网。

max_steps 一般设 3~5 步。太少了任务完不成,太大了浪费资源。3 步是一个不错的起点。

4.3 解析失败兜底

if action is None:
    break  # LLM 返回了无法解析的内容,退出

三次重试都失败,说明 LLM 卡住了。这时候不应该继续循环,而是直接退出。

三个终止条件的关系:

循环继续条件:
  ✅ AI 没说 done
  ✅ 步数 < max_steps
  ✅ 本步解析成功
  全部满足 → 继续
  任一不满足 → 退出

五、代码实现

5.1 状态类:AgentState

打开 agent/agent.py,首先找到新增的 AgentState 类:

class AgentState:
    """
    智能体状态追踪(第六课引入)

    跟踪智能体在循环中的进度:
    - 执行了多少步
    - 是否已完成
    - 每步的结果是什么
    """

    def __init__(self):
        self.steps: int = 0
        self.done: bool = False
        self.results: list[dict] = []

    def to_dict(self) -> dict:
        """将状态转换为字典(用于拼入 Prompt)"""
        return {
            "steps": self.steps,
            "done": self.done,
            "results": self.results[-3:],  # 只传最近3步,防止 prompt 过长
        }

    def add_result(self, result: dict) -> None:
        """记录一步的结果"""
        self.results.append(result)

    def increment_step(self) -> None:
        """步数 +1"""
        self.steps += 1

    def mark_done(self) -> None:
        """标记任务完成"""
        self.done = True

    def reset(self) -> None:
        """重置状态(开始新一轮循环前调用)"""
        self.steps = 0
        self.done = False
        self.results = []

注意一个细节:to_dict() 方法只返回最近 3 步的结果(self.results[-3:])。这是为了控制 prompt 长度——如果 Agent 跑了 10 步,把所有结果都塞进 prompt 会很长。只保留最近的 3 步,既能让 AI 看到上下文,又不会让 prompt 膨胀。

5.2 单步执行:agent_step()

def agent_step(self, user_input: str) -> dict | None:
    """
    执行智能体循环的一步:观察→决策→执行。

    第六课版本。

    Args:
        user_input: 用户输入或系统观察

    Returns:
        动作决策,如果步骤失败则返回 None
    """
    state_dict = self.state.to_dict()

    # 构建历史动作摘要
    history_summary = ""
    if state_dict.get("results"):
        history_lines = []
        for i, r in enumerate(state_dict["results"], 1):
            action = r.get("action", "?")
            reason = r.get("reason", "")
            history_lines.append(f"  步骤{i}: 动作={action}, 原因={reason}")
        history_summary = "\n之前的动作:\n" + "\n".join(history_lines)

    user_prompt = f"""你是一个智能体助手。你需要根据当前状态,决定下一步该做什么。

当前状态:步骤={state_dict.get('steps', 0)}, 已完成={state_dict.get('done', False)}
{history_summary}

可用动作:
- analyze:分析用户输入或已有结果
- research:深入研究某个方面
- summarize:总结已获得的信息
- answer:给出最终答案
- done:任务已完成,可以结束

规则:
1. 只返回有效的 JSON
2. 不要任何解释,不要 Markdown
3. 直接以 {{ 开头,以 }} 结尾
4. JSON 格式:{{"action": "动作名称", "reason": "选择该动作的原因"}}

用户输入:{user_input}

请返回 JSON:"""

    for attempt in range(3):
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": user_prompt},
            ],
            temperature=0.0,
        )

        text = response.choices[0].message.content
        parsed = extract_json_from_text(text)

        if parsed and "action" in parsed:
            if "reason" not in parsed:
                parsed["reason"] = f"执行动作: {parsed['action']}"
            self.state.increment_step()
            self.state.add_result(parsed)
            return parsed

    return None

这段代码你不会陌生——第三课的 JSON 提取、第四课的验证、第五课的 prompt 风格,全部复用:

JSON 输出 + extract_json_from_text() —— 第三课以来的标准操作。

重试 3 次 —— 老规矩了,LLM 有随机性,最多试三次。

验证关键字段 —— 检查 action 字段是否存在。始终验证模型输出。

temperature=0.0 —— 决策需要确定性,每一步该做什么不应该随机。

User Prompt 放动态内容 —— 可用动作列表和历史状态都是动态的,放 User Prompt。

唯一的新东西是 状态注入:在 prompt 里告诉模型当前执行到第几步、之前做了什么。这是让循环"有记忆"的关键。

5.3 循环执行:run_loop()

def run_loop(self, user_input: str, max_steps: int = 5) -> list[dict]:
    """
    运行多步智能体循环。

    Args:
        user_input: 初始用户输入
        max_steps: 最大执行步数(安全限制)

    Returns:
        每步的动作结果列表
    """
    self.state.reset()
    results = []

    print(f"🔄 启动智能体循环(最多 {max_steps} 步)")
    print(f"   用户输入:{user_input}\n")

    while not self.state.done and self.state.steps < max_steps:
        step_num = self.state.steps + 1
        print(f"--- 步骤 {step_num} ---")

        action = self.agent_step(user_input)

        if action:
            action_name = action.get("action", "?")
            reason = action.get("reason", "")
            print(f"   动作:{action_name}")
            print(f"   原因:{reason}")
            results.append(action)

            if action_name == "done":
                self.state.mark_done()
                print(f"   ✅ 智能体主动结束")
                break
        else:
            print(f"   ❌ 步骤失败,退出循环")
            break

        print()

    if not self.state.done and self.state.steps >= max_steps:
        print(f"⚠️ 达到最大步数 {max_steps},强制结束")

    print(f"\n📊 循环结束,共执行 {len(results)} 步")
    return results

仔细看这个循环的结构:

while not self.state.done and self.state.steps < max_steps:
    action = self.agent_step(user_input)

    if action:
        results.append(action)
        if action.get("action") == "done":
            self.state.mark_done()
    else:
        break

三个终止条件,一个都不能少:

终止条件 代码位置 作用
AI 主动完成 if action.get("action") == "done" 正常退出
最大步数 while self.state.steps < max_steps 安全限制
解析失败 if action: ... else: break 兜底

结果累积: 每步的 action 都被 append 到 results 列表里,最终整个循环返回一个完整的动作历史。


六、运行示例

6.1 基础运行

from agent.agent import Agent

agent = Agent(model="qwen2.5:7b")

results = agent.run_loop("帮我分析一下 Python 的优缺点", max_steps=3)

for i, result in enumerate(results, 1):
    action = result.get("action", "unknown")
    reason = result.get("reason", "无原因")
    print(f"  步骤{i}: [{action}] {reason}")

预期输出(类似):

🔄 启动智能体循环(最多 3 步)
   用户输入:帮我分析一下 Python 的优缺点

--- 步骤 1 ---
   动作:analyze
   原因:首先分析用户请求,识别需要讨论Python的优缺点

--- 步骤 2 ---
   动作:research
   原因:深入研究Python的具体优缺点细节

--- 步骤 3 ---
   动作:answer
   原因:已经收集了足够的信息,可以给出完整回答

📊 循环结束,共执行 3 步

注意一个现象:早期步骤中可能出现重复或相似的分析。这不是 Bug,而是 LLM 的正常行为——它在逐步完善自己的理解,可能在收敛到最终答案之前会重复探索。

这也正是 max_steps 的价值所在:即使 AI 在原地打转,步数到了也会被强制拉停。

6.2 更大的 max_steps

# 给 AI 更多步骤来完成复杂任务
results = agent.run_loop("帮我研究一下深度学习和传统机器学习的区别", max_steps=5)

你可能会观察到:

  • 步骤 1~2:分析用户需求,确定研究方向
  • 步骤 3~4:深入某个方面
  • 步骤 5:总结并给出答案(或者达到 max_steps 被强制结束)

6.3 交互模式

cd lesson06
python complete_example.py

会进入交互模式,你可以输入任意问题,观察 Agent 如何一步步处理。


七、与第五课的本质区别

把两课的流程放在一起对比,区别一目了然:

第五课(工具调用——单步):

用户输入 → AI 选择工具 → AI 提取参数 → 执行一次 → 返回结果

一次性。AI 选一个工具、带一组参数、执行一次、结束。每一步都是孤立的。

第六课(Agent Loop——多步):

用户输入
   ↓
步骤1:观察状态 → AI 决策 → 执行 → 更新状态
   ↓
步骤2:观察状态(含步骤1结果)→ AI 决策 → 执行 → 更新状态
   ↓
步骤3:观察状态(含步骤1-2结果)→ AI 决策 → 执行 → 更新状态
   ↓
……
   ↓
达到终止条件 → 结束

每一步都能看到之前所有步骤的结果。步骤之间有上下文、有累积、有推进。

第五课 第六课
执行次数 一次 多次(循环)
步骤间关系 无关系 前一步的结果影响下一步
状态追踪 有(steps、done、results)
终止条件 不需要 必须有(max_steps + done)
AI 的角色 “帮我选一个工具” “帮我规划并执行多步任务”

第五课的 AI 是"执行者"——你说一句,它做一件事。第六课的 AI 是"规划者"——它自己决定每一步该做什么,什么时候该停。


八、关键洞察

8.1 智能体不是聪明的 Prompt

这是本课最重要的一个洞察:

智能体不是聪明的提示词。智能体是带状态的循环。

你可以在一个单次调用的 prompt 里写"请分步骤思考",但那不叫 Agent。那只是让模型在输出里模拟多步推理。真正的 Agent 是你的代码在循环,每次循环调用一次 LLM,把结果存起来,再调用下一次。

魔力不在 prompt 里,而在循环里。

8.2 状态实现连续性

没有状态,每一步都是"失忆"的——AI 不知道自己做了什么、做到哪一步了。有了状态,步骤之间才能衔接。

这就是为什么 AgentState 虽然简单(就三个字段),但不可或缺。

8.3 终止条件是安全带

开车必须系安全带,写循环必须设终止条件。

max_steps 是你的安全带。不管 AI 的判断出了什么问题,步数到了就停。不要信任 AI 会主动说 “done”,你必须自己加限制。

8.4 简单先行,复杂后加

本课的循环刻意保持了简单——没有规划、没有记忆、没有复杂的推理链。先建立"循环 + 状态"的基础模式,后面的课程会在此基础上叠加更复杂的能力。

如果一上来就搞复杂的规划系统,反而容易迷失。先跑起来,再跑得更好。


九、常见问题

Q:Agent 在循环里反复做同一件事怎么办?

A:先检查 prompt 是否包含了历史状态信息。如果 AI 看不到之前做了什么,它当然会重复。确保 to_dict() 的结果被正确拼入了 prompt。如果状态信息已经有了但还是重复,可能是任务太模糊——试试给用户输入加更多约束,或者缩短 max_steps 减少浪费。

Q:循环永远不退出怎么办?

A:三重检查:① max_steps 条件是否在 while 里正确判断了?② self.state.increment_step() 是否在每次循环里被调用了?③ AI 输出的 done 动作是否被正确解析了?绝大多数"死循环"都是忘了调 increment_step()

Q:每一步的结果该传多少给下一步?

A:代码里用的是最近 3 步(self.results[-3:])。这是一个经验值——太少了 AI 看不到足够上下文,太多了 prompt 会很长。对于本地模型(7B),3~5 步的历史信息是比较合适的。如果你的模型上下文窗口更大,可以适当增加。

Q:max_steps 设多少合适?

A:取决于任务的复杂度。简单任务(分类、计算)3 步就够了。中等任务(分析、总结)5 步比较合适。复杂任务(研究、规划)可以到 10 步,但要注意 token 消耗。经验法则:从 3 开始,不够再加。

Q:状态只有 steps/done/results 够用吗?

A:对于本课的需求,完全够。但如果你想支持更复杂的场景,可以扩展状态。比如加一个 current_goal 字段记录当前子目标,或者加 errors 记录失败次数。第七课会加更强大的记忆系统。


十、下期预告

第七课:记忆——让 Agent 跨对话记住信息

本课的状态只在一次循环内有效。循环结束,状态清空。下一次对话,AI 又是"失忆"状态。

但真实的智能体需要跨对话记忆——它应该记住昨天你告诉它的偏好、上周它帮你做的分析、上个月你设定的目标。

下一课,我们给 Agent 加上持久化的记忆系统。这是从"能用的工具"到"有个性的助手"的关键一步。

敬请期待!


完整代码获取

本课涉及的完整代码包括:

  • AgentState 类——轻量级状态追踪
  • agent_step() 方法——单步执行逻辑
  • run_loop() 方法——多步循环引擎
  • complete_example.py——演示模式 + 交互模式

完整代码 参照 第一篇文章 最后


标签

#Python #AI Agent #LLM #智能体循环 #Ollama #Qwen #大模型 #手搓Agent


本文为《手搓 AI Agent 从 0 到 1》系列教程第 6 课

Logo

欢迎加入DeepSeek 技术社区。在这里,你可以找到志同道合的朋友,共同探索AI技术的奥秘。

更多推荐