14. 构建代理循环

在第 13 章中,我们学习了如何执行 LLM 的工具调用请求并返回结果。但那部分工作假设每个任务都可以在一次工具调用中完成。而在实践中,工具调用往往需要持续进行多轮,直到 LLM 收集到给出最终答案所需的全部信息为止。

考虑这样一个请求:"查找法国的人口,然后将其乘以二"。这个任务至少需要两次工具调用。你必须先查出人口数量——只有这样才能进行计算。第二次调用依赖于第一次的结果,所以无法在一次工具调用中完成。

在本章中,我们将把第 13 章的单次循环改造为一个循环。只要 LLM 请求工具调用,我们就持续执行它们——不断重复,直到 LLM 主动停止请求工具为止。接着我们会添加安全限制,防止循环无限运行,并介绍错误处理,让代理在工具失败时不会崩溃。

14.1) 从单次循环到循环

14.1.1) 代理循环是如何工作的?

正如引言中所看到的,那些下一步操作依赖于上一步结果的任务,往往无法通过一次工具调用完成。LLM 需要调用一个工具,检查结果,然后再次做出决策。这正是代理循环所做的事情,它分三个阶段工作:

思考(Think) —— LLM 读取到目前为止的对话内容,并决定接下来要做什么。如果需要工具,它会通过 tool_calls 请求一次工具调用。如果不需要,它就返回最终答案。
行动(Act) —— 执行 tool_calls 中指定的工具。
观察(Observe) —— 检查工具的执行结果,并将其以 ToolMessage 的形式添加到对话中。

代理循环会重复这三个阶段,直到 LLM 不再请求任何工具为止。这种模式也被称为 ReAct(Reason + Act,推理 + 行动),其核心思想是在推理和行动之间交替进行。

如果存在 tool_calls,就执行工具,将结果添加到对话中,然后再次调用 LLM。如果 tool_calls 为空,说明 LLM 已返回最终答案,循环终止。现在让我们把它写成代码。

14.1.2) 实现「思考-行动-观察」循环

让我们把上一节的「思考-行动-观察」循环写成代码。首先,我们准备好工具和模型。

python

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain.tools import tool
 
# 定义工具
@tool
def get_weather(city: str) -> str:
    """获取某个城市的当前天气。"""
    fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
    return fake_data.get(city, f"No weather data for {city}.")
 
@tool
def calculate(expression: str) -> str:
    """计算一个简单的算术表达式,例如 '3 * 21'。"""
    return str(eval(expression))  # 警告: eval() 存在安全风险。请勿在生产环境中使用。
 
# 绑定工具
tools = [get_weather, calculate]
llm = ChatOpenAI(model="gpt-5-mini")
llm_with_tools = llm.bind_tools(tools)
tool_map = {t.name: t for t in tools}

在第 13 章中,我们执行了一次工具就停止了。现在我们要不断重复,直到 LLM 停止请求工具调用为止。在一个 while True 循环中,我们调用 LLM,如果响应中包含 tool_calls,就执行这些工具并再次调用 LLM。如果没有 tool_calls,说明 LLM 已返回最终答案,于是我们退出循环。

python

def run_agent(user_input: str) -> str:
    """运行「思考-行动-观察」循环,直到 LLM 返回最终答案。"""
    messages = [
        SystemMessage(content="You are a helpful assistant."),
        HumanMessage(content=user_input),
    ]
 
    while True:
        # 思考: 请求 LLM 决定下一步操作
        print("THINK: Asking LLM to decide")
        ai_message = llm_with_tools.invoke(messages)
        messages.append(ai_message)
 
        # 退出条件: 没有 tool_calls 意味着这就是最终答案
        if not ai_message.tool_calls:
            return ai_message.content
 
        # 行动 + 观察: 执行请求的工具并将结果添加到对话中
        for tool_call in ai_message.tool_calls:
            selected_tool = tool_map[tool_call["name"]]
            print(f"ACT: calling '{tool_call['name']}', args={tool_call['args']}")
 
            tool_message = selected_tool.invoke(tool_call)
            print(f"OBSERVE: {tool_message.content}")
 
            messages.append(tool_message)

让我们把它和第 13 章的代码对比一下。在第 13 章中,执行完工具后,我们最后再调用一次 LLM 来获取答案。在第 14 章中,我们把这一相同的过程放进 while True 里,并在每次迭代中检查 tool_calls 来决定是否继续。这些构建块和第 13 章是一样的——我们只是把它们包进了一个循环里。

让我们运行它。

python

answer = run_agent("What's the weather in Tokyo, and is it warm enough for a walk?")
print(f'Final answer: {answer}')

输出:

THINK: Asking LLM to decide
ACT: calling 'get_weather', args={'city': 'Tokyo'}
OBSERVE: 18°C, cloudy
THINK: Asking LLM to decide
Final answer: 现在东京是 18°C(约 64°F),多云。
这个温度对大多数人来说通常温和舒适,适合散步。

输出展示了「思考 → 行动 → 观察 → 思考」的流程。在第一次迭代中,LLM 请求了一次 get_weather 调用;在第二次迭代中,它看到了天气结果并生成了最终答案。由于 tool_calls 为空,最终答案被返回,循环终止。

现在让我们测试引言中提到的那种依赖前一步结果的场景——只有在看到上一步的结果之后,下一次调用才能进行。

python

answer = run_agent("Get the temperature in Cairo, then multiply the number by 3.")
print(f'Final answer: {answer}')

输出:

THINK: Asking LLM to decide
ACT: calling 'get_weather', args={'city': 'Cairo'}
OBSERVE: 31°C, sunny
THINK: Asking LLM to decide
ACT: calling 'calculate', args={'expression': '31 * 3'}
OBSERVE: 93
THINK: Asking LLM to decide
Final answer: 开罗当前温度: 31°C。乘以 3 = 93。

这次循环运行了三次迭代。

第一次迭代 —— LLM 请求 get_weather("Cairo")。
第二次迭代 —— 在看到 "31°C, sunny" 结果后,LLM 请求 calculate("31 * 3")。它只有在看到温度后才能构造出这个表达式。
第三次迭代 —— 在看到 "31°C, sunny" 和 "93" 两个结果后,LLM 返回了最终答案。

14.2) 添加安全限制

我们上面构建的循环只有一个退出条件:当 LLM 返回时不带 tool_calls,我们就跳出循环。在正常情况下这已经足够了,但如果 LLM 永远不停止请求工具调用,会发生什么呢?

例如,如果某个工具总是返回含糊不清的结果,LLM 可能会不断调用它,期望得到更好的结果。由于循环是 while True,如果 LLM 不停止,程序也不会停止。API 调用成本会随着程序无限运行而不断累积。

最简单的解决办法是给循环能运行的次数设定一个上限。把 while True 替换为 for step in range(max_steps),那么无论 LLM 做什么,循环都保证会在 max_steps 次迭代后终止。

python

def run_agent(user_input: str, max_steps: int = 10) -> str:
    """在 max_steps 次迭代内运行「思考-行动-观察」循环。"""
    messages = [
        SystemMessage(content="You are a helpful assistant."),
        HumanMessage(content=user_input),
    ]
 
    for step in range(max_steps):
        # 思考
        ai_message = llm_with_tools.invoke(messages)
        messages.append(ai_message)
 
        # 退出条件: 没有 tool_calls 意味着这就是最终答案
        if not ai_message.tool_calls:
            return ai_message.content
 
        # 行动 + 观察
        for tool_call in ai_message.tool_calls:
            selected_tool = tool_map[tool_call["name"]]
            tool_message = selected_tool.invoke(tool_call)
            messages.append(tool_message)
 
    # 达到 max_steps: 循环结束但未得到最终答案
    return f"[Stopped after reaching max iterations ({max_steps})]"

与之前的代码相比,有两处变化。while True 变成了 for step in range(max_steps),并且在循环达到 max_steps 时添加了一个返回值。现在循环会以两种方式之一终止:LLM 主动返回最终答案(自然终止),或者达到 max_steps(安全终止)。

让我们验证一下安全限制是否真的有效。我们会创建一个永远不返回有用结果的工具,迫使 LLM 陷入一种它永远不停止请求工具调用的境地。

python

@tool
def unhelpful_search(query: str) -> str:
    """搜索信息。"""
    return "No results found. Try rephrasing your query."
 
llm_with_bad_tool = llm.bind_tools([unhelpful_search])
tool_map_bad = {unhelpful_search.name: unhelpful_search}
 
def run_agent_bad(user_input: str, max_steps: int = 5) -> str:
    messages = [
        SystemMessage(content=(
            "You must ALWAYS use the unhelpful_search tool to find information. "
            "You are NOT allowed to answer from your own knowledge. "
            "If the tool returns no results, you MUST rephrase and search again. "
            "Keep searching until you find the answer."
        )),
        HumanMessage(content=user_input),
    ]
 
    for step in range(max_steps):
        print(f"--- Step {step + 1} ---")
        ai_message = llm_with_bad_tool.invoke(messages)
        messages.append(ai_message)
 
        if not ai_message.tool_calls:
            return ai_message.content
 
        for tool_call in ai_message.tool_calls:
            selected_tool = tool_map_bad[tool_call["name"]]
            tool_message = selected_tool.invoke(tool_call)
            print(f"ACT: '{tool_call['name']}' → {tool_message.content}")
            messages.append(tool_message)
 
    return f"[Stopped after reaching max iterations ({max_steps})]"
 
answer = run_agent_bad("What is the population of France?")
print(f"\nFinal answer: {answer}")

输出:

--- Step 1 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
--- Step 2 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
--- Step 3 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
--- Step 4 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
--- Step 5 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
 
Final answer: [Stopped after reaching max iterations (5)]

如果没有 max_steps,这个循环会永远运行下去。多亏了 max_steps=5,它在五次迭代后被强制终止了。

max_steps 的合适取值取决于你的代理的复杂程度。取值太低,复杂任务会被过早中断。取值太高,行为异常的代理在被停止之前会累积大量成本。15–25 是一个常见的起点;可根据你的实际工作负载进行调整。

14.3) 在循环中处理工具错误

我们在 14.1 和 14.2 中构建的循环都假设工具总是执行成功。但如果某个工具抛出了异常,会发生什么呢?当前的代码没有异常处理,所以如果工具执行过程中发生异常,整个代理就会崩溃。

在第 12 章中,我们学习了如何在工具内部使用 try/except 来捕获异常,并将错误消息作为字符串返回。如果工具是这样构建的,那就没有问题。但并非每个工具都在内部处理错误。那些调用外部库或 API 的工具可能会抛出意料之外的异常。

为了防范这种情况,在循环层面同样处理错误是个好主意。方法很直接:把工具执行包裹在 try/except 中,如果发生异常,就把错误消息放进 ToolMessage 并传给 LLM。LLM 可以读取这条错误消息,然后用修正后的参数重试,或者选择一种不同的方法。这被称为自我修正(self-correction)。

python

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from langchain.tools import tool
 
# 定义工具
@tool
def calculate(expression: str) -> str:
    """计算一个简单的算术表达式,例如 '3 * 21'。"""
    return str(eval(expression))  # 警告: eval() 存在安全风险。请勿在生产环境中使用。
 
# 绑定工具
tools = [calculate]
llm = ChatOpenAI(model="gpt-5-mini")
llm_with_tools = llm.bind_tools(tools)
tool_map = {t.name: t for t in tools}
 
def run_agent(user_input: str, max_steps: int = 10) -> str:
    """将工具错误传给 LLM 的代理循环,从而实现自我修正。"""
    messages = [
        SystemMessage(content="You are a helpful assistant."),
        HumanMessage(content=user_input),
    ]
 
    for step in range(max_steps):
        # 思考
        print("THINK: Asking LLM to decide")
        ai_message = llm_with_tools.invoke(messages)
        messages.append(ai_message)
 
        if not ai_message.tool_calls:
            return ai_message.content
 
        # 行动 + 观察
        for tool_call in ai_message.tool_calls:
            selected_tool = tool_map[tool_call["name"]]
            try:
                print(f"ACT: calling '{tool_call['name']}', args={tool_call['args']}")
                tool_message = selected_tool.invoke(tool_call)
                print(f"OBSERVE: {tool_message.content}")
            except Exception as e:
                print(f"OBSERVE: Error - {e}")
                tool_message = ToolMessage(
                    content=f"Error: {e}",
                    tool_call_id=tool_call["id"],
                )
            messages.append(tool_message)
 
    return f"[Stopped after reaching max iterations ({max_steps})]"

与 14.2 的代码相比,唯一的变化就是 try/except。如果工具抛出异常,错误消息会被放进 ToolMessage 并添加到对话中。请注意,即使是失败的调用,也必须有一个带有匹配 tool_call_id 的 ToolMessage。LLM 会在下一次迭代中看到这个错误,并决定接下来要做什么。

让我们验证一下自我修正是否有效。我们会通过让 calculate 工具执行除以零来触发一个异常。

python

answer = run_agent("Use the calculator tool to compute 10 / 0")
print(f"Final answer: {answer}")

输出:

THINK: Asking LLM to decide
ACT: calling 'calculate', args={'expression': '10 / 0'}
OBSERVE: Error - division by zero
THINK: Asking LLM to decide
Final answer: 我使用了计算器工具,它返回了一个错误: "division by zero"。
 
解释: 10 / 0 在普通算术中是未定义的,因此它无法产生一个有限的数字。
 
您是否希望我:
- 计算单侧极限,
- 显示 IEEE-754 浮点结果,
- 或者计算一个不同的表达式?

在第一次迭代中,LLM 请求了 calculate("10 / 0"),于是抛出了一个 ZeroDivisionError。try/except 捕获了这个异常并把错误消息传给了 LLM。在第二次迭代中,LLM 看到了错误消息,返回了一个最终答案,解释除以零是不可能的。如果没有 try/except,程序会在第一个 ZeroDivisionError 处崩溃。