循环控制与错误恢复

第 6 章的循环能跑，但太「脆」：模型陷入死循环会烧钱、工具抛错会让整个任务中断、网络抖动会导致整轮失败。真实智能体必须有保险丝。本章把循环主体抽成可复用的 run_agent(system_prompt, tools, messages)，一次装好五种控制：步骤上限、死循环检测、工具报错恢复、重试退避、用户中断。这份 agent.py 是第 9 章起的公共运行器。

1. 五种保险一览

保险对付的问题手段
max_steps模型无限调工具步骤到顶即停，附道歉
死循环检测反复执行相同动作统计 (工具,参数) 签名，触顶回灌提示
工具报错恢复参数错误、工具内部异常错误以 tool 消息回填，模型自行修正
重试退避网络抖动、限流指数退避 + 随机抖动重发
用户中断不想等了Ctrl+C 捕获，保留已有上下文
另加一条隐含保险：检查 finish_reason == "length"，回复被 max_tokens 截断时提示模型「继续」而不是误当最终答案。

2. 完整代码 agent.py

"""agent.py —— 可控的智能体运行器（第 9 章起直接复用本模块）。

用法：
    from agent import run_agent
    from tools import TOOL_SCHEMAS
    messages = [{"role": "user", "content": "北京天气如何？"}]
    messages = run_agent(SYSTEM_PROMPT, TOOL_SCHEMAS, messages)
    print(messages[-1]["content"])   # 最终回答
"""
import json
import random
import time
from typing import Any, Dict, List, Optional

from llm_client import LLMClient, get_message
from tools import TOOL_SCHEMAS, call_tool

llm = LLMClient()

SYSTEM = "你是一个可靠的助手，用简体中文回答；工具报错时请依据错误信息修正参数或换方案重试。"


def assistant_to_dict(msg: Any) -> Dict[str, Any]:
    """assistant 消息转字典：有 tool_calls 时必须原样带回。"""
    item: Dict[str, Any] = {"role": "assistant", "content": msg.content or ""}
    if msg.tool_calls:
        item["tool_calls"] = [
            {"id": tc.id, "type": "function",
             "function": {"name": tc.function.name,
                          "arguments": tc.function.arguments}}
            for tc in msg.tool_calls
        ]
    return item


def chat_with_backoff(messages: List[dict],
                      tools: Optional[List[dict]] = None,
                      max_attempts: int = 4,
                      base_wait: float = 1.0) -> Any:
    """指数退避 + 随机抖动。llm.chat 自带 2 次简单重试，这里再兜底一层。"""
    for attempt in range(1, max_attempts + 1):
        try:
            return llm.chat(messages, tools=tools)
        except RuntimeError as e:
            if attempt >= max_attempts:
                raise
            wait = base_wait * (2 ** (attempt - 1)) + random.uniform(0, 0.5)
            print(f"[重试 {attempt + 1}/{max_attempts}] {wait:.1f} 秒后重发：{e}")
            time.sleep(wait)
    raise RuntimeError("重试耗尽")   # 防御性兜底，实际不会走到


def run_agent(system_prompt: str, tools: Optional[List[dict]],
              messages: List[dict], max_steps: int = 10,
              repeat_limit: int = 3) -> List[dict]:
    """执行一轮智能体任务，把本轮产生的消息写回 messages 并返回。

    - system_prompt：系统提示词
    - tools：工具 Schema 列表（不需要工具传 None）
    - messages：本轮对话（不含 system），末尾应是最近的 user 消息
    - max_steps：最大迭代步数
    - repeat_limit：相同动作重复多少次判为死循环
    返回后 messages[-1] 为最终回答（assistant 消息）。
    """
    conv: List[dict] = [{"role": "system", "content": system_prompt}, *messages]
    counts: Dict[str, int] = {}      # 动作签名 -> 出现次数，用于死循环检测

    for step in range(1, max_steps + 1):
        print(f"\n===== 第 {step}/{max_steps} 步 =====")
        try:
            resp = chat_with_backoff(conv, tools=tools)
        except KeyboardInterrupt:    # 保险 5：用户 Ctrl+C
            print("用户中断：已停止本轮，保留已有上下文。")
            break
        except RuntimeError as e:
            print(f"请求最终失败：{e}")
            conv.append({"role": "assistant",
                         "content": f"抱歉，请求多次失败未能完成：{e}"})
            break

        msg = get_message(resp)
        reason = resp.choices[0].finish_reason
        print("模型：", msg.content or "（发起工具调用）")

        if not msg.tool_calls:
            conv.append({"role": "assistant", "content": msg.content or ""})
            if reason == "length":   # 隐性保险：输出被截断，要求继续
                print("（回复被截断，请求模型继续）")
                conv.append({"role": "user", "content": "（上一条回复被截断，请继续）"})
                continue
            break                    # 正常结束：拿到最终答案

        conv.append(assistant_to_dict(msg))
        for tc in msg.tool_calls:
            name = tc.function.name
            try:
                args = json.loads(tc.function.arguments or "{}")
            except json.JSONDecodeError:
                args = {}
            key = f"{name}|{json.dumps(args, sort_keys=True)}"
            counts[key] = counts.get(key, 0) + 1

            if counts[key] >= repeat_limit + 1:   # 保险 2：提示过一次仍重复，终止
                result = "终止信号：相同动作重复过多。请停止调用工具，直接总结已有信息作答。"
            elif counts[key] == repeat_limit:     # 保险 2：首次触顶，回灌提示换思路
                result = (f"提示：你已重复执行 {name}{args} {repeat_limit} 次，疑似死循环。"
                          "请换一种思路，或直接用已有信息作答。")
            else:
                result = call_tool(name, args)    # 保险 3：正常执行
                if result.startswith("错误"):     # 工具报错也回填，不中断整个任务
                    result += "（请依据错误信息修正参数，或换一种方案）"
            print(f"  <- {name} 返回：{result}")
            conv.append({"role": "tool", "tool_call_id": tc.id, "content": result})
    else:                                         # 保险 1：步骤耗尽仍未结束
        print(f"达到步骤上限 {max_steps}，停止。")
        conv.append({"role": "assistant",
                     "content": f"抱歉，尝试 {max_steps} 步后仍未完成，请换一种问法。"})

    messages[:] = [m for m in conv if m["role"] != "system"]   # 写回（去掉 system）
    return messages


if __name__ == "__main__":
    question = "现在是几点？北京天气怎么样？顺便帮我算一下 123+456 等于多少。"
    messages = [{"role": "user", "content": question}]
    messages = run_agent(SYSTEM, TOOL_SCHEMAS, messages, max_steps=8)
    print("\n===== 最终回答 =====")
    print(messages[-1]["content"])

3. 运行与轨迹

python agent.py

轨迹与第 6 章相似（输出随模型与时间变化）：模型在第 1 步并行调用 get_current_time、get_weather、add，第 2 步汇总成最终答案；若某步工具参数写错，你会看到该步打印 错误：调用 ... 参数不正确...，随后模型在下一步自行修正——这就是「错误恢复」，整轮不会中断。

4. 各保险怎么工作

  • 步骤上限：for 循环天然带计数，走完 max_steps 仍未 break 就进 else 分支收尾。max_steps 是预算概念：既防死循环，也防单轮烧太多钱。
  • 死循环检测：以「工具名 + 排序后的参数 JSON」为动作签名计数。首次触顶（第 repeat_limit 次）回灌「疑似死循环」提示、不真执行；模型仍重复则改发终止信号，逼它收尾。相同参数连调 3 次误伤概率很低，阈值可调。
  • 工具报错恢复：call_tool 把所有异常都转成「错误：」开头字符串（第 5 章约定），这里原样作为 tool 消息回填并附一句提示。模型看到错误描述后通常会修正参数再试——报错不是任务的终点，而是模型的输入。
  • 重试退避：chat_with_backoff 在 llm.chat 内置重试之上，按 1s、2s、4s… 指数退避并叠加 0~0.5s 随机抖动，避免多个请求同时重试把服务打崩。超时由 .env 的 LLM_TIMEOUT 控制，属于请求级保险。
  • 用户中断：循环主体包 try/except KeyboardInterrupt，终端里 Ctrl+C 立刻停手，已产生的对话仍写回 messages，方便下次继续。

5. 复用方式

run_agent 的输入输出都是 messages 列表，天然支持连续对话与后续章节复用：

from agent import run_agent, SYSTEM
from tools import TOOL_SCHEMAS

history = [{"role": "user", "content": "我叫小林"}]
run_agent(SYSTEM, None, history)          # 不需要工具的闲聊轮
history.append({"role": "user", "content": "北京天气如何？"})
run_agent(SYSTEM, TOOL_SCHEMAS, history)  # 需要工具的轮次
print(history[-1]["content"])             # 上次回答仍记得“小林”

要换动作范式，只需替换 run_agent 内部「拿 tool_calls」一段为第 7 章的文本解析；要接记忆、HTTP 工具、流式输出，都在不改循环骨架的前提下往 messages / tools 里加东西即可。 小结：run_agent 用 max_steps 上限、动作签名计数、错误回填、指数退避、Ctrl+C 捕获五道保险把循环管住，并把「对话上下文」作为唯一进出参数，因此天然可复用、可续聊。到本章为止，一套带工具的智能体内核已经齐活，第 9 章起开始给它接真实世界的工具。

笔记加载中…

标签：#Agent实战

返回知识库