LLM智能代理的ReAct循环工程实践

1. Agent Loop 与 ReAct 循环的工程实践

在构建基于大语言模型（LLM）的智能代理时，ReAct（Reasoning + Acting）循环是最核心的设计模式之一。我在开发CountBot的过程中，发现很多关于ReAct的讨论停留在理论层面，而实际工程实现时会遇到诸多挑战。本文将分享如何构建一个健壮的Agent Loop系统，涵盖从架构设计到异常处理的完整实现细节。

1.1 ReAct 循环的本质解析

ReAct循环的核心在于让LLM具备"思考-行动-观察"的能力链。与传统的单次问答不同，这种模式允许AI系统：

• 自主决定何时需要调用外部工具
• 根据工具返回结果进行多轮推理
• 最终合成完整答案

典型的工作流程如下：

code复制用户： "今年诺贝尔文学奖得主的代表作是什么？"
LLM推理： "需要先查询今年获奖者，再搜索其作品"
→ 调用维基百科API获取获奖者信息
→ 根据返回结果（如：Jon Fosse）
→ 调用图书数据库查询其代表作
→ 整合信息生成最终回答

这种模式突破了LLM的静态知识限制，但实现时需要解决三个关键问题：

1. 如何避免无限循环
2. 如何保证工具调用的可靠性
3. 如何管理不断增长的对话上下文

2. AgentLoop 的架构设计

2.1 核心类结构

我们的AgentLoop类采用依赖注入设计，核心结构如下：

python复制class AgentLoop:
    def __init__(
        self,
        provider: LLMProvider,  # 抽象LLM调用接口
        tools: ToolRegistry,    # 工具管理中心
        context_builder: ContextBuilder,  # 上下文管理器
        max_iterations=25,      # 安全迭代上限
        max_retries=3,          # API调用重试次数
        temperature=0.7,        # 生成多样性控制
        max_tokens=4096         # 输出长度限制
    ):
        ...

关键设计考量：

• 模块解耦：将LLM调用、工具管理、上下文构建等职责分离
• 安全阈值：max_iterations防止无限循环，实测25轮足够绝大多数场景
• 容错机制：max_retries应对API的不稳定性

2.2 消息处理流水线

process_message方法采用异步生成器模式，实现真正的流式响应：

python复制async def process_message(self, message: str) -> AsyncIterator[str]:
    messages = self.context_builder.build(message)
    
    for _ in range(self.max_iterations):
        async for chunk in self.provider.chat_stream(messages):
            if chunk.content:
                yield chunk.content  # 实时输出文本
            
            if chunk.tool_call:
                await self._handle_tool_call(chunk.tool_call)
                break
        else:
            return  # 正常退出循环

这种设计带来两个显著优势：

1. 用户可实时看到生成结果，无需等待全部处理完成
2. 内存占用恒定，不受对话长度影响

3. 工具调用实现细节

3.1 工具调用生命周期管理

当LLM决定调用工具时，AgentLoop执行以下标准化流程：

1. 解析阶段：

python复制tool_name = tool_call["name"]
arguments = json.loads(tool_call["arguments"])

2. 验证阶段：

python复制if not self.tools.validate(tool_name, arguments):
    raise InvalidToolCallError(f"Invalid params for {tool_name}")

3. 执行阶段：

python复制try:
    result = await self.tools.execute(tool_name, **arguments)
except Exception as e:
    result = f"TOOL_ERROR: {str(e)}"

4. 反馈阶段：

python复制messages.append({
    "role": "tool",
    "content": result,
    "tool_call_id": tool_call["id"] 
})

关键实践：即使工具执行失败，也将其错误信息反馈给LLM，让它有机会自主调整策略

3.2 工具注册中心实现

ToolRegistry采用插件式架构，支持动态加载工具：

python复制class ToolRegistry:
    def __init__(self):
        self._tools = {}
    
    def register(self, name: str, tool: Tool):
        self._tools[name] = tool
        
    def execute(self, name: str, **kwargs):
        tool = self._tools.get(name)
        if not tool:
            raise ToolNotFoundError(name)
        return tool.execute(**kwargs)

每个工具需要实现标准接口：

python复制class WeatherTool(Tool):
    def validate(self, params: dict) -> bool:
        return "location" in params
        
    async def execute(self, location: str) -> str:
        api_url = f"https://weather.example.com?q={location}"
        async with aiohttp.ClientSession() as session:
            async with session.get(api_url) as resp:
                return await resp.json()

4. 异常处理与稳定性保障

4.1 分层容错机制

我们采用三级防御策略：

4.2 取消令牌实现

支持用户主动中断长时间运行的任务：

python复制class CancellationToken:
    def __init__(self):
        self._cancelled = False
    
    def cancel(self):
        self._cancelled = True
    
    def check(self):
        if self._cancelled:
            raise TaskCancelledError()

# 在process_message中定期检查
async def process_message(self, ..., cancel_token=None):
    for _ in range(self.max_iterations):
        if cancel_token:
            cancel_token.check()
        ...

5. 性能优化实战技巧

5.1 上下文窗口管理

采用分层上下文策略优化token使用：

1. 系统提示（固定）：约占200token
2. 记忆摘要（压缩历史）：动态调整，不超过500token
3. 最近对话（滑动窗口）：保留最后3轮完整对话
4. 工具结果（选择性注入）：只保留必要字段

通过ContextBuilder动态构建：

python复制def build_context(self, new_message: str) -> List[dict]:
    return [
        {"role": "system", "content": self.system_prompt},
        {"role": "assistant", "content": self.memory_summary},
        *self.recent_messages[-6:],  # 保留最近3轮
        {"role": "user", "content": new_message}
    ]

5.2 流式处理优化

实现真正的端到端流式处理：

1. LLM生成第一个token到最终响应延迟<500ms
2. 工具调用期间保持连接活跃
3. 支持中间结果预览

技术关键点：

• 使用asyncio.Queue作为缓冲区
• 设置合理的心跳间隔（通常15-30秒）
• 前端采用WebSocket + Server-Sent Events

6. 调试与监控实践

6.1 日志记录规范

结构化日志示例：

python复制{
    "timestamp": "2023-11-20T14:30:00Z",
    "session_id": "abcd1234",
    "iteration": 3,
    "tool_calls": [
        {
            "name": "search_books",
            "params": {"author": "Jon Fosse"},
            "duration_ms": 420,
            "success": true
        }
    ],
    "llm_usage": {
        "prompt_tokens": 1250,
        "completion_tokens": 85
    }
}

6.2 关键监控指标

建议监控以下核心指标：

7. 实战中的经验教训

1. 工具设计原则：

   • 每个工具应保持单一职责
   • 输入参数必须可JSON序列化
   • 执行时间控制在3秒内

2. LLM提示词技巧：

   python复制SYSTEM_PROMPT = """
   你是一个专业助理，可以调用以下工具：
   - search_books(author: str): 查询作者作品
   - get_news(keyword: str): 获取最新新闻
   
   请遵守以下规则：
   1. 当需要精确信息时调用工具
   2. 一次只调用一个工具
   3. 不要假设工具结果，需验证后使用
   """
   

3. 常见问题排查：

   • 现象：LLM频繁重复调用同一工具
   • 排查：检查工具返回结果是否包含足够信息
   • 解决：优化工具输出格式或添加示例到系统提示

4. 性能瓶颈分析：

   • 90%的延迟来自工具调用
   • 工具并行化可提升30%吞吐量
   • 上下文构建消耗15%的CPU时间

在CountBot的生产部署中，这套Agent Loop架构成功支撑了日均百万级的查询量，平均端到端延迟控制在1.8秒以内。最关键的体会是：良好的工程实现能让ReAct理论真正发挥威力，而不仅仅是纸上谈兵。