模块5: 构建智能体循环

每个AI智能体的核心都是一个循环。智能体观察当前状态（用户输入、工具结果、记忆），通过将上下文发送给LLM进行推理来思考，然后通过调用工具或返回最终响应来行动。这个循环重复直到任务完成。

这个模式被称为ReAct循环（推理+行动），最初在Yao等人2022年的论文中描述。它是每个现代智能体框架使用的相同模式——LangChain、CrewAI、OpenAI Agents SDK和Claude的工具使用——无论其具体API如何。

以下是伪代码形式的流程：

# The universal agent loop pattern
messages = [user_input]

while True:
    # THINK: Send context to LLM
    response = llm.generate(messages)

    # DECIDE: Does the LLM want to use a tool?
    if response.wants_tool_call:
        # ACT: Execute the tool
        tool_result = execute_tool(response.tool_call)

        # OBSERVE: Add result to context for next iteration
        messages.append(response)         # LLM's reasoning
        messages.append(tool_result)      # Tool's output
        continue                          # Loop back to THINK

    else:
        # DONE: LLM has a final answer
        return response.text

关键洞察是LLM自己决定何时停止。不是开发者硬编码"调用工具A，然后调用工具B，然后响应"。LLM根据任务和中间结果动态选择行动。这使得智能体具有灵活性——同一个循环可以处理"天气怎么样？"（一次工具调用）或"研究并总结前5名竞争对手"（多次工具调用和推理步骤）。

API级别的循环

OpenAI和Anthropic都通过响应对象来信号工具使用。模式是相同的：

让我们使用Anthropic SDK构建一个最小但完整的智能体。这个智能体可以查询天气——一个简单的例子，但它展示了使用真实API调用的完整观察-思考-行动循环。

第1步：定义工具

首先，定义智能体可以使用的工具。每个工具需要名称、描述和输入模式。描述至关重要——它告诉LLM何时使用该工具：

import anthropic

client = anthropic.Anthropic()

# Define the tools available to the agent
tools = [
    {
        "name": "get_weather",
        "description": "Get the current weather for a city. Use this when the user asks about weather, temperature, or conditions in a specific location.",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "The city name, e.g. 'London' or 'Tokyo'"
                }
            },
            "required": ["city"]
        }
    }
]

第2步：实现工具

在真实应用中，此函数会调用天气API。目前我们使用一个桩函数：

def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Execute a tool and return the result as a string."""
    if tool_name == "get_weather":
        city = tool_input["city"]
        # In production, call a real weather API here
        return f"Current weather in {city}: 22°C, partly cloudy, humidity 65%"
    else:
        return f"Error: Unknown tool '{tool_name}'"

第3步：智能体循环

现在是将一切联系在一起的核心循环：

def run_agent(user_message: str, max_steps: int = 10) -> str:
    """Run the agent loop until a final answer is produced."""
    print(f"\n{'='*50}")
    print(f"User: {user_message}")
    print(f"{'='*50}")

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

    for step in range(max_steps):
        print(f"\n--- Step {step + 1} ---")

        # THINK: Send context to the LLM
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system="You are a helpful assistant with access to tools.",
            tools=tools,
            messages=messages
        )

        print(f"Stop reason: {response.stop_reason}")

        # CHECK: Does the LLM want to call a tool?
        if response.stop_reason == "tool_use":
            # Find the tool_use block in the response
            tool_block = next(
                b for b in response.content if b.type == "tool_use"
            )
            print(f"Tool call: {tool_block.name}({tool_block.input})")

            # ACT: Execute the tool
            result = execute_tool(tool_block.name, tool_block.input)
            print(f"Tool result: {result}")

            # OBSERVE: Add both the LLM response and tool result to history
            messages.append({"role": "assistant", "content": response.content})
            messages.append({
                "role": "user",
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": tool_block.id,
                    "content": result
                }]
            })
            continue  # Back to THINK

        # DONE: LLM produced a final text response
        final_text = next(
            b.text for b in response.content if b.type == "text"
        )
        print(f"\nFinal answer: {final_text}")
        return final_text

    return "Error: Max steps reached without a final answer."


# Run it!
run_agent("What's the weather like in Tokyo?")

预期输出

==================================================
User: What's the weather like in Tokyo?
==================================================

--- Step 1 ---
Stop reason: tool_use
Tool call: get_weather({'city': 'Tokyo'})
Tool result: Current weather in Tokyo: 22°C, partly cloudy, humidity 65%

--- Step 2 ---
Stop reason: end_turn

Final answer: The current weather in Tokyo is 22°C with partly cloudy skies
and 65% humidity.

最小智能体 — OpenAI版本

使用OpenAI API的相同模式，以便你比较语法差异：

from openai import OpenAI
import json

client = OpenAI()

# OpenAI tool format wraps each tool in a "function" type
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather for a city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name"}
                },
                "required": ["city"]
            }
        }
    }
]

def run_agent_openai(user_message: str, max_steps: int = 10) -> str:
    messages = [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": user_message}
    ]

    for step in range(max_steps):
        response = client.chat.completions.create(
            model="gpt-4o",
            tools=tools,
            messages=messages
        )
        choice = response.choices[0]

        if choice.finish_reason == "tool_calls":
            tool_call = choice.message.tool_calls[0]
            args = json.loads(tool_call.function.arguments)
            result = execute_tool(tool_call.function.name, args)

            # OpenAI requires appending the assistant message first
            messages.append(choice.message)
            # Then the tool result with matching tool_call_id
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })
            continue

        return choice.message.content

    return "Max steps reached."

缺乏适当安全措施的智能体循环可能会失控——发起数十次API调用，消耗你的预算，并产生无意义的结果。安全机制不是可选的；它们是智能体设计的核心部分。

基本安全措施

在代码中实现安全措施

import time

def run_safe_agent(user_message: str, max_steps: int = 10,
                   max_tokens: int = 50000, timeout: int = 60) -> str:
    """Agent loop with comprehensive safety measures."""
    messages = [{"role": "user", "content": user_message}]
    total_tokens = 0
    start_time = time.time()

    for step in range(max_steps):
        # Safety check: timeout
        elapsed = time.time() - start_time
        if elapsed > timeout:
            print(f"TIMEOUT: Agent exceeded {timeout}s limit at step {step + 1}")
            return "Error: Agent timed out."

        # THINK
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system="You are a helpful assistant.",
            tools=tools,
            messages=messages
        )

        # Safety check: token budget
        step_tokens = response.usage.input_tokens + response.usage.output_tokens
        total_tokens += step_tokens
        print(f"Step {step + 1}: {step_tokens} tokens (total: {total_tokens})")

        if total_tokens > max_tokens:
            print(f"BUDGET: Exceeded {max_tokens} token limit")
            return "Error: Token budget exceeded."

        # Normal loop logic continues...
        if response.stop_reason == "tool_use":
            tool_block = next(
                b for b in response.content if b.type == "tool_use"
            )
            result = execute_tool(tool_block.name, tool_block.input)
            messages.append({"role": "assistant", "content": response.content})
            messages.append({
                "role": "user",
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": tool_block.id,
                    "content": result
                }]
            })
            continue

        final = next(b.text for b in response.content if b.type == "text")
        print(f"\nCompleted in {step + 1} steps, {total_tokens} tokens, "
              f"{time.time() - start_time:.1f}s")
        return final

    return "Error: Max steps reached."

检测无限循环

一种更微妙的失败模式是智能体不断用相同的输入调用相同的工具，得到相同的结果，却从不取得进展。你可以通过跟踪最近的工具调用来检测这种情况：

def detect_loop(messages: list, lookback: int = 4) -> bool:
    """Check if the agent is stuck calling the same tool repeatedly."""
    recent_tool_calls = []
    for msg in messages[-lookback * 2:]:  # Check last N exchanges
        if isinstance(msg.get("content"), list):
            for block in msg["content"]:
                if isinstance(block, dict) and block.get("type") == "tool_result":
                    recent_tool_calls.append(block.get("content", ""))

    # If all recent tool results are identical, we're probably looping
    if len(recent_tool_calls) >= 3 and len(set(recent_tool_calls)) == 1:
        return True
    return False