这次我们来看一个很值得花时间梳理的架构方向类PI-Agent复杂智能体架构设计。标题里的核心线索是“模型与智能体从单次调用到 Loop”。很多做 LLM 应用的开发者最初都是写一个prompt → 接口 → 结果的单次调用但真正到了 Agent 阶段问题会突然变复杂模型要自己决定下一步做什么、要调哪些工具、要记住前面几轮的结果、要在异常时重新尝试。这一整套能力不是靠换一个大模型就能解决的而是需要在模型外面包一层完整的运行架构。这篇文章会把“单次调用”到“Agent Loop”这条演进路线拆开从设计思路到可运行的代码再到测试、接口化、批量任务、性能观察和排错清单一次讲透。文章适用的读者很明确正在做 Agent 开发、RAG 应用、自动化工作流、LLM 工具调用的工程同学。如果你已经在用 Dify、Coze 这类智能体平台但想知道底层这类“类 PI-Agent”的 Loop 架构到底是怎么设计和实现的这篇也很合适。文章默认你具备 Python 基础能运行 Python 脚本并理解大模型 API 的基本调用方式。整体的阅读路径是先搞清楚为什么单次调用不够用再理解 Loop 的核心设计然后跟着代码落地最后补上工程化测试、批量任务、性能观察和常见问题排查。1. 核心能力速览能力项说明架构类型类 PI-Agent 复杂智能体架构强调规划Planning与执行Implement闭环核心机制从单次模型调用演进到 Agent Loop 循环支持工具调用、状态累积、迭代决策关键组件模型接入层、Agent 运行循环、工具注册与执行、上下文管理、应用接口层模型兼容性兼容 OpenAI 格式的接口服务可接入云端模型也可通过 vLLM、Ollama 等接本地模型语言与技术栈Python 3.10依赖 openai、FastAPI、pydantic 等是否支持 CPU取决于底层模型架构本身与 CPU/GPU 无关是否支持 API支持可使用 FastAPI 将 Agent 封装为 HTTP 接口是否支持批量任务支持可基于线程池或任务队列做并发批量处理是否支持多智能体可扩展支持 Planner、Executor、Reflector 等多角色协作适合场景本地 Agent 原型验证、工具调用系统设计、多步任务自动化、接口化 Agent 服务部署复杂度中低依赖干净适合从单体脚本逐步演进为服务这些能力项里最核心的设计目标不是“调用一个大模型”而是构建一个可持续循环的 Agent 运行时。模型只负责推理和决策工具负责真实执行Loop 负责让“推理-执行-观察”反复发生直到任务完成。2. 适用场景与使用边界这类架构适合以下几个典型场景。第一类是本地 Agent 原型验证。很多团队在正式引入商业化 Agent 平台之前会先用 Python 写一个简单 Loop把模型接入、工具调用、上下文维护跑通验证这个思路在自己的业务数据上是否有效。第二类是工具调用型任务。例如让 Agent 自己决定调用数据库查询、调用搜索接口、调用内部 API这类任务天然需要“模型生成调用参数 — 程序执行 — 把结果回填给模型”的循环。第三类是自动化工作流。比如批量处理文档、自动生成结构化报告、定时巡检业务数据Agent 在这些场景里可以根据中间结果动态调整后续步骤比固定流程更灵活。但也有不适合的情况。首先是超低延迟场景Agent Loop 一次完整执行往往需要多次模型调用端到端耗时会明显高于单次请求如果业务要求 300ms 内返回这类架构就不合适。其次是成本敏感场景每一轮循环都会消耗 Token循环越长成本越高如果任务本身很简单强行套 Agent Loop 会放大费用。第三是强合规场景如果 Agent 要自主操作外部系统例如发送邮件、提交订单、修改配置必须有严格的人工审批和权限管控不能直接开放给模型自由调用。从合规和伦理边界来看使用 Agent 架构时必须注意几个底线。一是数据安全不要把未脱敏的隐私数据直接丢给云端模型优先考虑本地部署或者在传输链路中做加密和审计。二是授权边界Agent 调用外部工具前要确认操作是否在授权范围内尤其是涉及人脸、声音、版权素材、用户隐私的场景。三是内容审核模型在循环中如果生成不当内容或者工具执行返回了异常数据系统需要有兜底过滤机制而不是无限重试。四是可追溯所有模型输入、输出、工具调用记录都应留存日志便于事后审计和责任界定。3. 为什么需要从单次调用到 Loop模型不是智能体很多初学者会有一个误解模型调用次数多了就是智能体。其实不是。单次调用和 Agent Loop 之间差的不是调用次数而是“决策回路”。先看单次调用。它的流程是固定的用户输入程序拼接 Prompt请求模型接口拿到输出展示给用户。整个流程只有一个“推理点”模型没有机会验证自己的输出是否正确没有机会修正错误也没有机会去获取模型参数之外的实时信息。比如你问模型“今天北京天气怎么样”如果模型没有联网工具它只能根据训练数据猜测或者直接告诉你它不知道。它在回答之前无法主动发起一次天气查询这就是单次调用的结构性局限。单次调用的局限可以归纳为三点无工具能力模型只能用自己的参数知识回答无法调用外部 API、数据库或代码执行器。无状态累积每次请求是独立的模型看不到自己前面几轮的中间判断长任务拆解能力很弱。无自适应路径所有任务都走同一条 Prompt 路径模型无法根据中间结果决定下一步走哪条分支。Agent Loop 解决的正是这三个问题。它的运行逻辑是一个循环模型基于当前状态做一次推理决策如果决策结果是调用某个工具系统就执行这个工具把结果作为新的消息回填给模型模型再继续决策直到模型认为任务已经完成输出最终答案。这个循环在学术和工程上通常被称为 ReAct 模式也就是推理Reasoning加行动Action的交替进行。类 PI-Agent 的思想在这里面很集中把“规划”和“执行”放到同一个循环里模型负责规划工具层负责执行每执行一步就把观察结果带回规划器从而形成闭环。用一句直白的话说单次调用是“问一句、答一句”Agent Loop 是“边干边想干完再想直到干完”。4. 类 PI-Agent 架构整体设计这里说的类 PI-Agent可以把 PI 理解为 Planning-Intelligence也可以理解为 Plan-Implement 的闭环。无论哪种解释核心都是一致的Agent 不是单个模型函数而是一个包含多层的软件系统。从分层角度看我建议把架构划分为四层。第一层是模型接入层。它负责统一封装底层大模型接口屏蔽不同模型服务商的差异。无论你接的是云端模型还是本地模型向上层提供的都应该是一套兼容 OpenAI Chat Completions 格式的接口。这一层需要处理超时、重试、流式输出、模型名配置等细节。第二层是 Agent 运行层。这是整个架构的核心包含 Agent Loop 主循环、状态管理、迭代次数控制、终止条件判断。Agent 的所有决策都发生在这里。运行层需要回答几个关键问题什么时候继续循环什么时候结束循环工具调用失败后怎么办上下文太长怎么压缩。第三层是工具与服务层。Agent 需要执行的真实动作都在这一层。工具可以是查询数据库、调用外部 REST API、执行 Python 代码、读写文件、发送通知等。每个工具需要把名称、参数 Schema、执行函数、错误处理封装在一起注册到工具注册表里。设计上要保证工具是原子的、可观测的、可重试的。第四层是应用接入层。它对外提供 API、批量任务入口、日志监控和人工审批接口。实际业务应用通过这一层与 Agent 交互不需要关心内部循环逻辑。这四层之间的数据流是这样的用户请求进入应用接入层传递给 Agent 运行层运行层准备好初始消息调用模型接入层获取一次决策如果决策包含工具调用请求运行层把请求交给工具层执行工具层返回执行结果运行层把结果追加到消息列表再调用模型接入层如此循环直到模型给出最终回答。关键组件可以用下面的表格汇总组件职责设计要点Model Client封装模型接口统一超时、重试、模型名配置Tool Registry管理工具注册与执行名称唯一、参数校验、错误统一处理Context Manager管理消息历史控制消息长度、窗口裁剪、摘要压缩Loop Controller控制循环与终止最大迭代次数、结束条件、异常退出Task Queue批量任务调度并发数、失败重试、任务状态记录Audit Logger行为审计日志记录输入输出、工具调用、耗时 Token5. 环境准备与前置条件部署这套架构不需要很重的环境。整个系统是纯 Python 服务可以在 Linux 服务器上跑也可以在 Windows 或 macOS 上本地调试。推荐使用 Python 3.10 及以上版本这样可以比较舒服地使用新版类型注解和match语法。依赖管理建议使用venv或conda创建独立虚拟环境避免和系统 Python 环境冲突。基础依赖包括依赖库用途openai调用 OpenAI 兼容接口接本地或云端模型fastapi将 Agent 封装成 HTTP API 服务uvicorn运行 FastAPI 应用的 ASGI 服务器pydantic请求参数校验和工具参数 Schema 管理httpx同步或异步 HTTP 请求用于工具调用模型服务的准备有两种方式。如果有云端 API Key直接使用服务商提供的base_url和模型名即可。如果希望本地部署可以先用 Ollama 或 vLLM 起一个 OpenAI 兼容服务。例如本地使用 vLLM 启动时服务默认会暴露http://127.0.0.1:8000/v1的 OpenAI 兼容接口这时在代码里把base_url指向它api_key填一个占位符就能跑通。安装依赖的命令如下# 创建虚拟环境 python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate # 安装依赖 pip install openai fastapi uvicorn pydantic httpx环境就绪后建议先用一个最小脚本确认模型接口能通再进入架构代码编写。把网络、模型服务、API Key 这类外部依赖先验证清楚后续排查 Agent 问题时更容易定位是模型问题还是循环逻辑问题。6. 核心代码实现从单次调用到 Agent Loop这一章是实战主体。我会按照“先写单次调用 → 再封装工具注册表 → 再实现 Agent Loop → 再补上下文管理”的顺序展开。这样你能清楚看到每一步在做什么以及演进到 Loop 后解决了什么问题。6.1 单次调用基础代码先写最基础的模型调用。这里假设你接的是一个 OpenAI 兼容接口可能是云端服务也可能是本地 vLLM 或 Ollama。核心代码如下from openai import OpenAI # 如果是本地 OpenAI 兼容服务api_key 可以填任意占位符 client OpenAI( api_keyyour-api-key, base_urlhttp://127.0.0.1:8000/v1, ) def single_call(prompt: str, model: str your-model-name) - str: 单次模型调用不做任何循环和工具处理。 response client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], temperature0.7, ) return response.choices[0].message.content if __name__ __main__: print(single_call(用一句话介绍什么是 Agent Loop))这段代码很简单但它暴露了前面说的三个问题模型无法调用工具、无法记住多轮决策、无法根据结果动态调整下一步。6.2 工具注册表实现要让模型具备工具能力第一步是先建立一个工具注册表。工具注册表负责登记工具的元信息、参数 Schema 和实际执行函数。模型需要工具时只会拿到工具的 Schema程序执行工具时才真正调用对应的 Python 函数。import json from typing import Callable class ToolRegistry: 工具注册表管理 Agent 可调用的所有外部工具。 def __init__(self): self._functions: dict[str, Callable] {} self._schemas: list[dict] [] def register(self, name: str, func: Callable, schema: dict): 注册一个工具。 Args: name: 工具名称Agent 通过该名称调用工具。 func: 实际执行函数。 schema: OpenAI Function Calling 格式的参数 Schema。 if name in self._functions: raise ValueError(ftool already registered: {name}) self._functions[name] func self._schemas.append({ type: function, function: { name: name, description: schema.get(description, ), parameters: schema.get(parameters, {}), }, }) def get_schemas(self) - list[dict]: 返回模型可用的工具 Schema 列表。 return self._schemas def execute(self, name: str, arguments: dict): 执行指定名称的工具。 if name not in self._functions: raise KeyError(ftool not found: {name}) func self._functions[name] # 这里可以根据需要加入参数校验、日志、耗时统计 return func(**arguments)注册一个天气查询模拟工具和日期工具from datetime import datetime def get_current_date() - str: 返回当前日期用于测试工具调用。 return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_weather(city: str) - dict: 模拟天气查询工具。实际项目中应替换为真实 API 调用。 demo_data { 北京: {weather: 晴, temperature: 20}, 上海: {weather: 多云, temperature: 23}, } return demo_data.get(city, {weather: 未知, temperature: None}) registry ToolRegistry() registry.register( nameget_current_date, funcget_current_date, schema{ description: 获取当前日期和时间, parameters: { type: object, properties: {}, }, }, ) registry.register( nameget_weather, funcget_weather, schema{ description: 查询指定城市的天气信息, parameters: { type: object, properties: { city: {type: string, description: 城市名称如北京、上海} }, required: [city], }, }, )这套设计是通用的实际项目中你可以把get_weather替换为数据库查询、内部 API 调用或文件操作结构不需要改动。6.3 Agent Loop 核心循环有了工具注册表下一步就是实现 Agent Loop。这是整个架构的心脏。Loop 的逻辑是把用户消息交给模型 — 模型返回决策 — 如果决策包含工具调用就执行工具把结果回填给模型 — 继续循环 — 直到模型不再请求工具输出最终结果。import json from openai import OpenAI class AgentLoop: 类 PI-Agent 的核心循环规划-执行-观察直到模型给出最终回答。 def __init__( self, client: OpenAI, model: str, registry: ToolRegistry, max_iterations: int 10, ): self.client client self.model model self.registry registry self.max_iterations max_iterations def run(self, user_prompt: str) - str: messages [{role: user, content: user_prompt}] for step in range(1, self.max_iterations 1): print(f[Agent] step {step}/{self.max_iterations}) response self.client.chat.completions.create( modelself.model, messagesmessages, toolsself.registry.get_schemas(), tool_choiceauto, ) message response.choices[0].message # 模型没有要求调用工具说明任务完成输出最终回答 if not message.tool_calls: return message.content or # 先把模型的工具调用请求追加到消息列表保持对话上下文完整 messages.append(message) # 逐个执行模型请求的工具 for tool_call in message.tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments or {}) print(f[Agent] call tool: {tool_name} args{tool_args}) try: result self.registry.execute(tool_name, tool_args) except Exception as e: result {error: str(e)} # 工具执行结果以 tool 角色消息回填给模型 messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(result, ensure_asciiFalse), }) raise RuntimeError( fagent loop reached max_iterations{self.max_iterations} )这个循环是理解整个架构的关键。注意几个细节模型可能一次请求多个工具所以内层要用for循环逐个执行。工具调用请求本身要以 assistant 角色消息追加到messages中工具结果用 tool 角色且带tool_call_id与之关联。这部分必须符合 OpenAI 工具调用的消息格式。如果工具执行抛异常不要把异常直接丢弃而是转成{error: ...}回填给模型让模型自己判断下一步怎么处理这比程序直接中断更符合 Agent 的容错逻辑。max_iterations是必须的保险丝。没有它一个失控的循环会无限调用模型既烧 Token 又卡住服务。运行示例client OpenAI( api_keyyour-api-key, base_urlhttp://127.0.0.1:8000/v1, ) agent AgentLoop( clientclient, modelyour-model-name, registryregistry, max_iterations5, ) if __name__ __main__: answer agent.run(现在几点了顺便查一下北京和上海的天气。) print(最终回答:, answer)一个正常执行流程会是模型第一次决策调用get_current_date和get_weather系统执行工具后把结果回填模型看到时间和天气数据后输出最终回答。整个过程在日志里可以看到两次模型调用中间夹着两次工具执行。6.4 上下文管理与压缩Agent 每循环一轮消息列表就会增加至少两条消息。如果任务复杂循环十几轮消息列表会非常长最终触及模型的上下文窗口限制。所以上下文管理不能省。最简单的实现是固定窗口保留 system 消息只保留最近 N 条消息。更高级的做法是摘要压缩当消息超过阈值时把旧消息交给模型生成摘要用摘要替代原始消息继续循环。def compress_messages( messages: list[dict], client: OpenAI, model: str, keep_last: int 6, ) - list[dict]: 简单的窗口压缩保留首条 system 消息和最近 keep_last 条消息。 Args: messages: 当前完整消息列表。 client: 模型客户端。 model: 用于压缩的模型名称。 keep_last: 保留最近的消息数量。 Returns: 压缩后的消息列表。 if len(messages) keep_last 1: return messages system_messages [m for m in messages if m.get(role) system] recent_messages messages[-keep_last:] # 将较早消息做摘要 history_to_summarize messages[1:-keep_last] if history_to_summarize: summary_prompt ( 请压缩下面这段对话历史保留关键信息和已经完成的工具调用结果\n json.dumps(history_to_summarize, ensure_asciiFalse) ) summary_response client.chat.completions.create( modelmodel, messages[{role: user, content: summary_prompt}], ) summary_text summary_response.choices[0].message.content return system_messages [ {role: system, content: f对话历史摘要{summary_text}}, ] recent_messages return system_messages recent_messages这个实现适合原型阶段。生产环境建议使用独立的消息数据库或 Redis 保存全量历史Loop 里只维护当前窗口。另外模型上下文长度不同压缩阈值需要按实际模型规格调整。7. 功能测试与效果验证架构代码写完后的第一件事不是直接上业务而是做一组功能验证。对 Agent Loop 来说测试的核心不是单次回答质量而是循环决策的稳定性和工具调用的正确性。我建议按以下矩阵来设计测试用例测试场景测试输入示例预期行为判断标准无工具调用“用一句话介绍你自己”模型直接回答不触发工具日志中无工具调用记录单工具调用“现在几点了”调用 get_current_date返回时间工具执行成功回答包含时间多工具并行调用“查北京和上海天气”一次决策调用多个工具日志显示多个 tool_call工具参数提取“查询城市北京”正确提取 city 参数工具收到的参数为北京工具返回异常工具内部抛异常错误信息回填模型模型修复或停止不崩溃有最终回答达到最大迭代构造无法完成的任务触发 max_iterations 抛错抛出 RuntimeError长任务上下文压缩模拟 20 轮历史压缩后仍能继续回答无上下文超限错误实际操作时建议先把日志打开观察每一步的模型决策。一个健康的 Agent Loop 应该满足模型每次工具调用都有明确目的工具结果正确回填最终回答依赖了工具返回的数据而不是模型自己编造。测试时最容易遇到的问题主要有这几类。第一是模型不支持 Function Calling。这时调用接口会报错建议换用支持工具调用的模型或者在 Prompt 中要求模型输出固定格式的 JSON再由程序解析。第二是工具参数解析失败模型生成的参数和 Schema 不匹配。此时可以在工具执行前加 pydantic 校验或者提高模型温度调低到 0.2 左右。第三是 Loop 陷入死循环反复调用同一个工具。这时需要增加重复调用检测比如同一工具连续调用超过 3 次就直接终止。8. 接口 API 与批量任务接入Loop 原型跑通后下一步就是把它封装成服务让别人可以调用。8.1 FastAPI 接口封装使用 FastAPI 把 Agent 包装成 HTTP 服务是最直接的方式。from fastapi import FastAPI from pydantic import BaseModel, Field app FastAPI(titleAgent Loop API) class AgentRequest(BaseModel): prompt: str Field(..., description用户输入) max_iterations: int Field(10, ge1, le50, description最大循环次数) class AgentResponse(BaseModel): result: str # 生产环境建议返回完整 trace 日志 trace: list[dict] | None None app.post(/agent/run, response_modelAgentResponse) def run_agent(request: AgentRequest): 同步执行一次 Agent Loop返回最终结果。 # 生产环境建议为每个请求创建独立 client或使用连接池 client OpenAI( api_keyyour-api-key, base_urlhttp://127.0.0.1:8000/v1, ) agent AgentLoop( clientclient, modelyour-model-name, registryregistry, max_iterationsrequest.max_iterations, ) result agent.run(request.prompt) return AgentResponse(resultresult)启动服务uvicorn main:app --host 127.0.0.1 --port 8000用 curl 验证接口curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d {prompt: 现在几点了, max_iterations: 5}Python 调用方式import requests response requests.post( http://127.0.0.1:8000/agent/run, json{prompt: 查一下北京和上海的天气, max_iterations: 5}, timeout120, ) print(response.json())这里需要特别注意一个问题同步接口的耗时等于整个 Agent Loop 的耗时可能是几秒也可能是一两分钟。调用方的timeout要设置得足够长。如果接口需要支撑实时页面建议改为异步任务模式接口先返回task_id客户端轮询任务状态。8.2 批量任务队列批量任务场景下不能依次同步调用效率太低。推荐做法是用线程池控制并发配合重试和日志。import logging from concurrent.futures import ThreadPoolExecutor, as_completed logger logging.getLogger(agent.batch) def run_single_with_retry( prompt: str, max_retries: int 3, max_iterations: int 10, ) - dict: 执行单个 Agent 任务带失败重试。 for attempt in range(1, max_retries 1): try: client OpenAI( api_keyyour-api-key, base_urlhttp://127.0.0.1:8000/v1, ) agent AgentLoop( clientclient, modelyour-model-name, registryregistry, max_iterationsmax_iterations, ) result agent.run(prompt) return {prompt: prompt, result: result, status: success} except Exception as e: logger.warning(attempt %d failed for prompt %s: %s, attempt, prompt, e) if attempt max_retries: return {prompt: prompt, error: str(e), status: failed} def batch_run( prompts: list[str], max_workers: int 4, max_retries: int 3, ) - list[dict]: 并发执行多个 Agent 任务。 results [] with ThreadPoolExecutor(max_workersmax_workers) as pool: future_map { pool.submit(run_single_with_retry, p, max_retries): p for p in prompts } for future in as_completed(future_map): try: results.append(future.result()) except Exception as e: prompt future_map[future] results.append({prompt: prompt, error: str(e), status: failed}) return results批量任务的工程化要点有四个。第一并发数要按模型服务的 QPS 限制调整并发太高会触发限流。第二每个任务要有唯一任务 ID日志中要能按任务 ID 串联模型调用和工具执行记录。第三失败重试要区分“模型接口错误”和“Agent 逻辑错误”接口超时可以重试但工具逻辑错误重试也没有意义。第四批量结果需要落盘或入库保存不能只打印到终端。9. 资源占用与性能观察Agent Loop 的性能模型和普通 API 调用完全不同。一次 Agent 任务会触发多次模型推理每次推理都消耗时间和 Token因此观察性能要同时关注三个维度端到端耗时、Token 消耗、工具执行耗时。端到端耗时主要由“模型推理次数 × 单次推理耗时”决定。假设一个任务要循环 4 轮每轮模型推理 2 秒那么仅模型推理就消耗约 8 秒这还不包括工具执行和网络传输。如果任务需要循环 10 次以上端到端耗时会很容易超过 30 秒。因此对 Loop 做耗时预算很重要。可以在 AgentLoop 里为每次模型调用和工具调用打点记录耗时最后输出统计信息。Token 消耗是更值得关注的成本因素。Agent Loop 的 Token 开销并不只是“每次调用的 Token 总和”还有一个容易忽略的部分是历史消息越长后续每轮推理的输入 Token 越大。这就会出现一个现象第 1 轮只消耗 500 Token但第 10 轮可能消耗 4000 Token因为前面 9 轮的工具结果都在输入里。要控制成本除了上下文压缩还需要控制循环深度并对单轮任务设置合理的 max_iterations。显存占用方面如果使用本地模型主要取决于模型服务进程。Agent 架构本身不会显著增加显存占用因为 Agent 服务只是做 HTTP 转发和消息拼接。但并发批量任务时如果每个任务都新建连接请求本地模型服务模型服务的显存会被推理并发度影响。建议批量任务并发数先设为 1 或 2观察模型服务响应时间再逐步调高。资源观察的基本方法是加计时日志。具体建议在每次模型调用前后记录耗时和 Token 使用量。在每次工具调用前后记录耗时和结果大小。在 AgentLoop.run 结束前汇总输出本轮循环的总耗时、总 Token、工具调用次数。批量任务中记录每个任务耗时用于识别慢任务。如果发现慢任务集中在工具调用阶段优先排查工具本身的性能如果慢任务分布在模型调用阶段优先考虑降低循环轮次、压缩上下文或升级模型服务配置。10. 多智能体扩展思路单个 Agent Loop 能解决一部分问题但业务复杂度上来后单一 Agent 会面临一个矛盾既要具备专业领域深度又要处理跨度很大的任务。这种情况下多智能体协作是更合理的架构方向。类 PI-Agent 的架构可以很自然地扩展为多智能体协作。最常见的分工模式包括三种角色Planner 负责把复杂任务拆解为子任务规划执行顺序然后把每个子任务分发给对应的执行智能体。Executor 负责执行具体子任务它内部可以是另一个 Agent Loop也可以只是一个工具调用。Reflector 负责检查执行结果判断是否达到预期如果没达到就反馈给 Planner 重新规划。这种设计的价值在于职责隔离和上下文隔离。每个子 Agent 只维护与自己任务相关的上下文不会被无关的历史消息干扰。相比单一超长 Agent Loop多智能体协作在复杂任务上往往更稳定也更容易调试和测试。多智能体之间的通信方式通常有两种。一种是消息传递Agent 之间通过任务结果队列或消息总线传递结构化数据另一种是共享黑板模式多个 Agent 读写共享的工作区状态。原型阶段建议先用消息传递结构清晰出了问题容易追踪。生产环境如果需要多个智能体并发处理子任务再引入消息队列如 Redis Stream 或 RabbitMQ。从单次调用到单 Agent Loop再到多 Agent 协作这条演进路径本质上是在不断增加系统的“决策粒度”。每次增加决策点系统的灵活性都会上升但随之而来的是成本、延迟和调试难度的同步上升。架构设计的目标不是把系统做到最复杂而是找到业务收益和技术成本之间的平衡点。11. 常见问题与排查方法问题现象可能原因排查方式解决方案接口报 model 不存在模型名配置错误查看模型服务启动日志确认可用的模型列表修改 model 参数为正确的模型名模型不触发工具调用模型不支持 Function Calling或工具 Schema 不合法打印 tools 参数检查 Schema 格式换一个已知支持工具调用的大模型测试切换兼容模型或在 Prompt 中要求固定 JSON 输出工具参数解析失败模型生成的参数不符合 Schema打印 tool_call.function.arguments比对 Schema增加 pydantic 参数校验降低 temperatureLoop 陷入重复调用工具返回结果不满足模型预期模型反复尝试同一工具查看日志中同一工具调用次数增加重复调用检测连续 3 次相同则终止并总结上下文超限错误消息列表超过模型最大上下文长度查看报错信息的 token 数使用 compress_messages 压缩限制 max_iterations接口响应超时Agent Loop 循环次数过多或模型推理慢查看耗时日志定位高耗时阶段降低 max_iterations改用异步任务接口批量任务部分失败模型服务限流或工具偶发异常查看失败任务日志区分错误类型增加重试和退避降低并发数工具执行后模型回答不准确工具结果回填格式不对查看 tool 角色消息内容确保 content 是 JSON 字符串且包含关键结果显存不足本地模型服务并发推理过高查看模型服务日志和 nvidia-smi降低并发数或换更小的模型排查的基本原则是先看日志再看消息内容最后才怀疑模型能力。Agent Loop 的每一步决策、每次工具调用都应该有日志没有日志时排查会非常困难。12. 最佳实践与合规边界架构层面最值得记住的几个实践如下。第一第一次跑通永远用小参数。max_iterations 设成 3工具只注册一个模型用最快的版本。先确认 Loop 链路通再逐步增加工具和任务难度。第二维护一套最小可运行配置包括最小依赖、最小模型名、最小工具集。这样后续升级或定位问题时永远有一个可以回退的基线。第三工具调用必须可观测。每个工具的执行时间、输入参数、输出结果都要记录不然 Loop 出了一次诡异的结果你根本无法定位是从哪一步开始错的。第四批量任务必须加失败重试和结果落盘。没有重试的批量任务是脆弱的没有落盘的结果是不可审计的。第五接口服务要限制访问范围尤其在局域网或公网开放 Agent 服务时。Agent 具备工具调用能力如果被未授权的人调用等于把你的内部工具暴露给外部。必须增加身份认证和操作审批。合规层面Agent 涉及的工具调用如果作用于真实业务系统需要特别注意权限设计。建议把工具的权限收敛到最小可用范围例如数据库账号只读、外部接口只限定指定操作。涉及人脸、声音、版权素材、个人隐私数据的处理场景必须确认素材来源合法、用户已授权处理过程要有审计日志。涉及自动化操作例如自动发消息、自动下单、自动修改配置的必须设置人工审批节点不能让模型在循环中直接执行高风险操作。发布或商用前还要对 Agent 的输出做抽样复核确认不会生成违规或误导性内容。13. 总结整篇文章的核心就一句话模型是推理引擎智能体是围绕模型构建的循环系统。从单次调用到 Agent Loop变化的不是模型能力而是架构设计。你要在模型外面补上工具注册、上下文管理、循环控制、终止条件、批量调度、日志审计这些工程组件才能把一个“会聊天的模型”变成一个“能干活的任务执行系统”。如果看完这篇文章只打算做一件事建议先实现一个最简 Agent Loop一个工具、三轮回合、输出日志。先把循环和工具回填跑通你会对模型决策和工具执行之间的关系有非常直观的感受。在这个过程中最容易踩的坑是上下文管理没做就疯狂加工具导致消息列表爆炸后续所有轮次变慢变贵。正确顺序是先把短循环跑稳再加工具再加上下文压缩最后再考虑多智能体扩展。后续如果你想继续深入可以往这几个方向扩展一是把上下文压缩模块换成向量数据库长期记忆二是把线程池批处理换成 RabbitMQ 或 Redis Stream 任务队列三是增加人工审批组件把高风险工具操作纳入审批流四是尝试 Planner-Executor-Reflector 多智能体协作把本文的 Loop 作为 Executor 的内核。每一步扩展都以“可观测、可回退、可审计”为前提这套类 PI-Agent 架构就能从原型平滑长成生产级系统。 从单次调用到Agent Loop：类PI-Agent智能体架构解析
这次我们来看一个很值得花时间梳理的架构方向类PI-Agent复杂智能体架构设计。标题里的核心线索是“模型与智能体从单次调用到 Loop”。很多做 LLM 应用的开发者最初都是写一个prompt → 接口 → 结果的单次调用但真正到了 Agent 阶段问题会突然变复杂模型要自己决定下一步做什么、要调哪些工具、要记住前面几轮的结果、要在异常时重新尝试。这一整套能力不是靠换一个大模型就能解决的而是需要在模型外面包一层完整的运行架构。这篇文章会把“单次调用”到“Agent Loop”这条演进路线拆开从设计思路到可运行的代码再到测试、接口化、批量任务、性能观察和排错清单一次讲透。文章适用的读者很明确正在做 Agent 开发、RAG 应用、自动化工作流、LLM 工具调用的工程同学。如果你已经在用 Dify、Coze 这类智能体平台但想知道底层这类“类 PI-Agent”的 Loop 架构到底是怎么设计和实现的这篇也很合适。文章默认你具备 Python 基础能运行 Python 脚本并理解大模型 API 的基本调用方式。整体的阅读路径是先搞清楚为什么单次调用不够用再理解 Loop 的核心设计然后跟着代码落地最后补上工程化测试、批量任务、性能观察和常见问题排查。1. 核心能力速览能力项说明架构类型类 PI-Agent 复杂智能体架构强调规划Planning与执行Implement闭环核心机制从单次模型调用演进到 Agent Loop 循环支持工具调用、状态累积、迭代决策关键组件模型接入层、Agent 运行循环、工具注册与执行、上下文管理、应用接口层模型兼容性兼容 OpenAI 格式的接口服务可接入云端模型也可通过 vLLM、Ollama 等接本地模型语言与技术栈Python 3.10依赖 openai、FastAPI、pydantic 等是否支持 CPU取决于底层模型架构本身与 CPU/GPU 无关是否支持 API支持可使用 FastAPI 将 Agent 封装为 HTTP 接口是否支持批量任务支持可基于线程池或任务队列做并发批量处理是否支持多智能体可扩展支持 Planner、Executor、Reflector 等多角色协作适合场景本地 Agent 原型验证、工具调用系统设计、多步任务自动化、接口化 Agent 服务部署复杂度中低依赖干净适合从单体脚本逐步演进为服务这些能力项里最核心的设计目标不是“调用一个大模型”而是构建一个可持续循环的 Agent 运行时。模型只负责推理和决策工具负责真实执行Loop 负责让“推理-执行-观察”反复发生直到任务完成。2. 适用场景与使用边界这类架构适合以下几个典型场景。第一类是本地 Agent 原型验证。很多团队在正式引入商业化 Agent 平台之前会先用 Python 写一个简单 Loop把模型接入、工具调用、上下文维护跑通验证这个思路在自己的业务数据上是否有效。第二类是工具调用型任务。例如让 Agent 自己决定调用数据库查询、调用搜索接口、调用内部 API这类任务天然需要“模型生成调用参数 — 程序执行 — 把结果回填给模型”的循环。第三类是自动化工作流。比如批量处理文档、自动生成结构化报告、定时巡检业务数据Agent 在这些场景里可以根据中间结果动态调整后续步骤比固定流程更灵活。但也有不适合的情况。首先是超低延迟场景Agent Loop 一次完整执行往往需要多次模型调用端到端耗时会明显高于单次请求如果业务要求 300ms 内返回这类架构就不合适。其次是成本敏感场景每一轮循环都会消耗 Token循环越长成本越高如果任务本身很简单强行套 Agent Loop 会放大费用。第三是强合规场景如果 Agent 要自主操作外部系统例如发送邮件、提交订单、修改配置必须有严格的人工审批和权限管控不能直接开放给模型自由调用。从合规和伦理边界来看使用 Agent 架构时必须注意几个底线。一是数据安全不要把未脱敏的隐私数据直接丢给云端模型优先考虑本地部署或者在传输链路中做加密和审计。二是授权边界Agent 调用外部工具前要确认操作是否在授权范围内尤其是涉及人脸、声音、版权素材、用户隐私的场景。三是内容审核模型在循环中如果生成不当内容或者工具执行返回了异常数据系统需要有兜底过滤机制而不是无限重试。四是可追溯所有模型输入、输出、工具调用记录都应留存日志便于事后审计和责任界定。3. 为什么需要从单次调用到 Loop模型不是智能体很多初学者会有一个误解模型调用次数多了就是智能体。其实不是。单次调用和 Agent Loop 之间差的不是调用次数而是“决策回路”。先看单次调用。它的流程是固定的用户输入程序拼接 Prompt请求模型接口拿到输出展示给用户。整个流程只有一个“推理点”模型没有机会验证自己的输出是否正确没有机会修正错误也没有机会去获取模型参数之外的实时信息。比如你问模型“今天北京天气怎么样”如果模型没有联网工具它只能根据训练数据猜测或者直接告诉你它不知道。它在回答之前无法主动发起一次天气查询这就是单次调用的结构性局限。单次调用的局限可以归纳为三点无工具能力模型只能用自己的参数知识回答无法调用外部 API、数据库或代码执行器。无状态累积每次请求是独立的模型看不到自己前面几轮的中间判断长任务拆解能力很弱。无自适应路径所有任务都走同一条 Prompt 路径模型无法根据中间结果决定下一步走哪条分支。Agent Loop 解决的正是这三个问题。它的运行逻辑是一个循环模型基于当前状态做一次推理决策如果决策结果是调用某个工具系统就执行这个工具把结果作为新的消息回填给模型模型再继续决策直到模型认为任务已经完成输出最终答案。这个循环在学术和工程上通常被称为 ReAct 模式也就是推理Reasoning加行动Action的交替进行。类 PI-Agent 的思想在这里面很集中把“规划”和“执行”放到同一个循环里模型负责规划工具层负责执行每执行一步就把观察结果带回规划器从而形成闭环。用一句直白的话说单次调用是“问一句、答一句”Agent Loop 是“边干边想干完再想直到干完”。4. 类 PI-Agent 架构整体设计这里说的类 PI-Agent可以把 PI 理解为 Planning-Intelligence也可以理解为 Plan-Implement 的闭环。无论哪种解释核心都是一致的Agent 不是单个模型函数而是一个包含多层的软件系统。从分层角度看我建议把架构划分为四层。第一层是模型接入层。它负责统一封装底层大模型接口屏蔽不同模型服务商的差异。无论你接的是云端模型还是本地模型向上层提供的都应该是一套兼容 OpenAI Chat Completions 格式的接口。这一层需要处理超时、重试、流式输出、模型名配置等细节。第二层是 Agent 运行层。这是整个架构的核心包含 Agent Loop 主循环、状态管理、迭代次数控制、终止条件判断。Agent 的所有决策都发生在这里。运行层需要回答几个关键问题什么时候继续循环什么时候结束循环工具调用失败后怎么办上下文太长怎么压缩。第三层是工具与服务层。Agent 需要执行的真实动作都在这一层。工具可以是查询数据库、调用外部 REST API、执行 Python 代码、读写文件、发送通知等。每个工具需要把名称、参数 Schema、执行函数、错误处理封装在一起注册到工具注册表里。设计上要保证工具是原子的、可观测的、可重试的。第四层是应用接入层。它对外提供 API、批量任务入口、日志监控和人工审批接口。实际业务应用通过这一层与 Agent 交互不需要关心内部循环逻辑。这四层之间的数据流是这样的用户请求进入应用接入层传递给 Agent 运行层运行层准备好初始消息调用模型接入层获取一次决策如果决策包含工具调用请求运行层把请求交给工具层执行工具层返回执行结果运行层把结果追加到消息列表再调用模型接入层如此循环直到模型给出最终回答。关键组件可以用下面的表格汇总组件职责设计要点Model Client封装模型接口统一超时、重试、模型名配置Tool Registry管理工具注册与执行名称唯一、参数校验、错误统一处理Context Manager管理消息历史控制消息长度、窗口裁剪、摘要压缩Loop Controller控制循环与终止最大迭代次数、结束条件、异常退出Task Queue批量任务调度并发数、失败重试、任务状态记录Audit Logger行为审计日志记录输入输出、工具调用、耗时 Token5. 环境准备与前置条件部署这套架构不需要很重的环境。整个系统是纯 Python 服务可以在 Linux 服务器上跑也可以在 Windows 或 macOS 上本地调试。推荐使用 Python 3.10 及以上版本这样可以比较舒服地使用新版类型注解和match语法。依赖管理建议使用venv或conda创建独立虚拟环境避免和系统 Python 环境冲突。基础依赖包括依赖库用途openai调用 OpenAI 兼容接口接本地或云端模型fastapi将 Agent 封装成 HTTP API 服务uvicorn运行 FastAPI 应用的 ASGI 服务器pydantic请求参数校验和工具参数 Schema 管理httpx同步或异步 HTTP 请求用于工具调用模型服务的准备有两种方式。如果有云端 API Key直接使用服务商提供的base_url和模型名即可。如果希望本地部署可以先用 Ollama 或 vLLM 起一个 OpenAI 兼容服务。例如本地使用 vLLM 启动时服务默认会暴露http://127.0.0.1:8000/v1的 OpenAI 兼容接口这时在代码里把base_url指向它api_key填一个占位符就能跑通。安装依赖的命令如下# 创建虚拟环境 python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate # 安装依赖 pip install openai fastapi uvicorn pydantic httpx环境就绪后建议先用一个最小脚本确认模型接口能通再进入架构代码编写。把网络、模型服务、API Key 这类外部依赖先验证清楚后续排查 Agent 问题时更容易定位是模型问题还是循环逻辑问题。6. 核心代码实现从单次调用到 Agent Loop这一章是实战主体。我会按照“先写单次调用 → 再封装工具注册表 → 再实现 Agent Loop → 再补上下文管理”的顺序展开。这样你能清楚看到每一步在做什么以及演进到 Loop 后解决了什么问题。6.1 单次调用基础代码先写最基础的模型调用。这里假设你接的是一个 OpenAI 兼容接口可能是云端服务也可能是本地 vLLM 或 Ollama。核心代码如下from openai import OpenAI # 如果是本地 OpenAI 兼容服务api_key 可以填任意占位符 client OpenAI( api_keyyour-api-key, base_urlhttp://127.0.0.1:8000/v1, ) def single_call(prompt: str, model: str your-model-name) - str: 单次模型调用不做任何循环和工具处理。 response client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], temperature0.7, ) return response.choices[0].message.content if __name__ __main__: print(single_call(用一句话介绍什么是 Agent Loop))这段代码很简单但它暴露了前面说的三个问题模型无法调用工具、无法记住多轮决策、无法根据结果动态调整下一步。6.2 工具注册表实现要让模型具备工具能力第一步是先建立一个工具注册表。工具注册表负责登记工具的元信息、参数 Schema 和实际执行函数。模型需要工具时只会拿到工具的 Schema程序执行工具时才真正调用对应的 Python 函数。import json from typing import Callable class ToolRegistry: 工具注册表管理 Agent 可调用的所有外部工具。 def __init__(self): self._functions: dict[str, Callable] {} self._schemas: list[dict] [] def register(self, name: str, func: Callable, schema: dict): 注册一个工具。 Args: name: 工具名称Agent 通过该名称调用工具。 func: 实际执行函数。 schema: OpenAI Function Calling 格式的参数 Schema。 if name in self._functions: raise ValueError(ftool already registered: {name}) self._functions[name] func self._schemas.append({ type: function, function: { name: name, description: schema.get(description, ), parameters: schema.get(parameters, {}), }, }) def get_schemas(self) - list[dict]: 返回模型可用的工具 Schema 列表。 return self._schemas def execute(self, name: str, arguments: dict): 执行指定名称的工具。 if name not in self._functions: raise KeyError(ftool not found: {name}) func self._functions[name] # 这里可以根据需要加入参数校验、日志、耗时统计 return func(**arguments)注册一个天气查询模拟工具和日期工具from datetime import datetime def get_current_date() - str: 返回当前日期用于测试工具调用。 return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_weather(city: str) - dict: 模拟天气查询工具。实际项目中应替换为真实 API 调用。 demo_data { 北京: {weather: 晴, temperature: 20}, 上海: {weather: 多云, temperature: 23}, } return demo_data.get(city, {weather: 未知, temperature: None}) registry ToolRegistry() registry.register( nameget_current_date, funcget_current_date, schema{ description: 获取当前日期和时间, parameters: { type: object, properties: {}, }, }, ) registry.register( nameget_weather, funcget_weather, schema{ description: 查询指定城市的天气信息, parameters: { type: object, properties: { city: {type: string, description: 城市名称如北京、上海} }, required: [city], }, }, )这套设计是通用的实际项目中你可以把get_weather替换为数据库查询、内部 API 调用或文件操作结构不需要改动。6.3 Agent Loop 核心循环有了工具注册表下一步就是实现 Agent Loop。这是整个架构的心脏。Loop 的逻辑是把用户消息交给模型 — 模型返回决策 — 如果决策包含工具调用就执行工具把结果回填给模型 — 继续循环 — 直到模型不再请求工具输出最终结果。import json from openai import OpenAI class AgentLoop: 类 PI-Agent 的核心循环规划-执行-观察直到模型给出最终回答。 def __init__( self, client: OpenAI, model: str, registry: ToolRegistry, max_iterations: int 10, ): self.client client self.model model self.registry registry self.max_iterations max_iterations def run(self, user_prompt: str) - str: messages [{role: user, content: user_prompt}] for step in range(1, self.max_iterations 1): print(f[Agent] step {step}/{self.max_iterations}) response self.client.chat.completions.create( modelself.model, messagesmessages, toolsself.registry.get_schemas(), tool_choiceauto, ) message response.choices[0].message # 模型没有要求调用工具说明任务完成输出最终回答 if not message.tool_calls: return message.content or # 先把模型的工具调用请求追加到消息列表保持对话上下文完整 messages.append(message) # 逐个执行模型请求的工具 for tool_call in message.tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments or {}) print(f[Agent] call tool: {tool_name} args{tool_args}) try: result self.registry.execute(tool_name, tool_args) except Exception as e: result {error: str(e)} # 工具执行结果以 tool 角色消息回填给模型 messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(result, ensure_asciiFalse), }) raise RuntimeError( fagent loop reached max_iterations{self.max_iterations} )这个循环是理解整个架构的关键。注意几个细节模型可能一次请求多个工具所以内层要用for循环逐个执行。工具调用请求本身要以 assistant 角色消息追加到messages中工具结果用 tool 角色且带tool_call_id与之关联。这部分必须符合 OpenAI 工具调用的消息格式。如果工具执行抛异常不要把异常直接丢弃而是转成{error: ...}回填给模型让模型自己判断下一步怎么处理这比程序直接中断更符合 Agent 的容错逻辑。max_iterations是必须的保险丝。没有它一个失控的循环会无限调用模型既烧 Token 又卡住服务。运行示例client OpenAI( api_keyyour-api-key, base_urlhttp://127.0.0.1:8000/v1, ) agent AgentLoop( clientclient, modelyour-model-name, registryregistry, max_iterations5, ) if __name__ __main__: answer agent.run(现在几点了顺便查一下北京和上海的天气。) print(最终回答:, answer)一个正常执行流程会是模型第一次决策调用get_current_date和get_weather系统执行工具后把结果回填模型看到时间和天气数据后输出最终回答。整个过程在日志里可以看到两次模型调用中间夹着两次工具执行。6.4 上下文管理与压缩Agent 每循环一轮消息列表就会增加至少两条消息。如果任务复杂循环十几轮消息列表会非常长最终触及模型的上下文窗口限制。所以上下文管理不能省。最简单的实现是固定窗口保留 system 消息只保留最近 N 条消息。更高级的做法是摘要压缩当消息超过阈值时把旧消息交给模型生成摘要用摘要替代原始消息继续循环。def compress_messages( messages: list[dict], client: OpenAI, model: str, keep_last: int 6, ) - list[dict]: 简单的窗口压缩保留首条 system 消息和最近 keep_last 条消息。 Args: messages: 当前完整消息列表。 client: 模型客户端。 model: 用于压缩的模型名称。 keep_last: 保留最近的消息数量。 Returns: 压缩后的消息列表。 if len(messages) keep_last 1: return messages system_messages [m for m in messages if m.get(role) system] recent_messages messages[-keep_last:] # 将较早消息做摘要 history_to_summarize messages[1:-keep_last] if history_to_summarize: summary_prompt ( 请压缩下面这段对话历史保留关键信息和已经完成的工具调用结果\n json.dumps(history_to_summarize, ensure_asciiFalse) ) summary_response client.chat.completions.create( modelmodel, messages[{role: user, content: summary_prompt}], ) summary_text summary_response.choices[0].message.content return system_messages [ {role: system, content: f对话历史摘要{summary_text}}, ] recent_messages return system_messages recent_messages这个实现适合原型阶段。生产环境建议使用独立的消息数据库或 Redis 保存全量历史Loop 里只维护当前窗口。另外模型上下文长度不同压缩阈值需要按实际模型规格调整。7. 功能测试与效果验证架构代码写完后的第一件事不是直接上业务而是做一组功能验证。对 Agent Loop 来说测试的核心不是单次回答质量而是循环决策的稳定性和工具调用的正确性。我建议按以下矩阵来设计测试用例测试场景测试输入示例预期行为判断标准无工具调用“用一句话介绍你自己”模型直接回答不触发工具日志中无工具调用记录单工具调用“现在几点了”调用 get_current_date返回时间工具执行成功回答包含时间多工具并行调用“查北京和上海天气”一次决策调用多个工具日志显示多个 tool_call工具参数提取“查询城市北京”正确提取 city 参数工具收到的参数为北京工具返回异常工具内部抛异常错误信息回填模型模型修复或停止不崩溃有最终回答达到最大迭代构造无法完成的任务触发 max_iterations 抛错抛出 RuntimeError长任务上下文压缩模拟 20 轮历史压缩后仍能继续回答无上下文超限错误实际操作时建议先把日志打开观察每一步的模型决策。一个健康的 Agent Loop 应该满足模型每次工具调用都有明确目的工具结果正确回填最终回答依赖了工具返回的数据而不是模型自己编造。测试时最容易遇到的问题主要有这几类。第一是模型不支持 Function Calling。这时调用接口会报错建议换用支持工具调用的模型或者在 Prompt 中要求模型输出固定格式的 JSON再由程序解析。第二是工具参数解析失败模型生成的参数和 Schema 不匹配。此时可以在工具执行前加 pydantic 校验或者提高模型温度调低到 0.2 左右。第三是 Loop 陷入死循环反复调用同一个工具。这时需要增加重复调用检测比如同一工具连续调用超过 3 次就直接终止。8. 接口 API 与批量任务接入Loop 原型跑通后下一步就是把它封装成服务让别人可以调用。8.1 FastAPI 接口封装使用 FastAPI 把 Agent 包装成 HTTP 服务是最直接的方式。from fastapi import FastAPI from pydantic import BaseModel, Field app FastAPI(titleAgent Loop API) class AgentRequest(BaseModel): prompt: str Field(..., description用户输入) max_iterations: int Field(10, ge1, le50, description最大循环次数) class AgentResponse(BaseModel): result: str # 生产环境建议返回完整 trace 日志 trace: list[dict] | None None app.post(/agent/run, response_modelAgentResponse) def run_agent(request: AgentRequest): 同步执行一次 Agent Loop返回最终结果。 # 生产环境建议为每个请求创建独立 client或使用连接池 client OpenAI( api_keyyour-api-key, base_urlhttp://127.0.0.1:8000/v1, ) agent AgentLoop( clientclient, modelyour-model-name, registryregistry, max_iterationsrequest.max_iterations, ) result agent.run(request.prompt) return AgentResponse(resultresult)启动服务uvicorn main:app --host 127.0.0.1 --port 8000用 curl 验证接口curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d {prompt: 现在几点了, max_iterations: 5}Python 调用方式import requests response requests.post( http://127.0.0.1:8000/agent/run, json{prompt: 查一下北京和上海的天气, max_iterations: 5}, timeout120, ) print(response.json())这里需要特别注意一个问题同步接口的耗时等于整个 Agent Loop 的耗时可能是几秒也可能是一两分钟。调用方的timeout要设置得足够长。如果接口需要支撑实时页面建议改为异步任务模式接口先返回task_id客户端轮询任务状态。8.2 批量任务队列批量任务场景下不能依次同步调用效率太低。推荐做法是用线程池控制并发配合重试和日志。import logging from concurrent.futures import ThreadPoolExecutor, as_completed logger logging.getLogger(agent.batch) def run_single_with_retry( prompt: str, max_retries: int 3, max_iterations: int 10, ) - dict: 执行单个 Agent 任务带失败重试。 for attempt in range(1, max_retries 1): try: client OpenAI( api_keyyour-api-key, base_urlhttp://127.0.0.1:8000/v1, ) agent AgentLoop( clientclient, modelyour-model-name, registryregistry, max_iterationsmax_iterations, ) result agent.run(prompt) return {prompt: prompt, result: result, status: success} except Exception as e: logger.warning(attempt %d failed for prompt %s: %s, attempt, prompt, e) if attempt max_retries: return {prompt: prompt, error: str(e), status: failed} def batch_run( prompts: list[str], max_workers: int 4, max_retries: int 3, ) - list[dict]: 并发执行多个 Agent 任务。 results [] with ThreadPoolExecutor(max_workersmax_workers) as pool: future_map { pool.submit(run_single_with_retry, p, max_retries): p for p in prompts } for future in as_completed(future_map): try: results.append(future.result()) except Exception as e: prompt future_map[future] results.append({prompt: prompt, error: str(e), status: failed}) return results批量任务的工程化要点有四个。第一并发数要按模型服务的 QPS 限制调整并发太高会触发限流。第二每个任务要有唯一任务 ID日志中要能按任务 ID 串联模型调用和工具执行记录。第三失败重试要区分“模型接口错误”和“Agent 逻辑错误”接口超时可以重试但工具逻辑错误重试也没有意义。第四批量结果需要落盘或入库保存不能只打印到终端。9. 资源占用与性能观察Agent Loop 的性能模型和普通 API 调用完全不同。一次 Agent 任务会触发多次模型推理每次推理都消耗时间和 Token因此观察性能要同时关注三个维度端到端耗时、Token 消耗、工具执行耗时。端到端耗时主要由“模型推理次数 × 单次推理耗时”决定。假设一个任务要循环 4 轮每轮模型推理 2 秒那么仅模型推理就消耗约 8 秒这还不包括工具执行和网络传输。如果任务需要循环 10 次以上端到端耗时会很容易超过 30 秒。因此对 Loop 做耗时预算很重要。可以在 AgentLoop 里为每次模型调用和工具调用打点记录耗时最后输出统计信息。Token 消耗是更值得关注的成本因素。Agent Loop 的 Token 开销并不只是“每次调用的 Token 总和”还有一个容易忽略的部分是历史消息越长后续每轮推理的输入 Token 越大。这就会出现一个现象第 1 轮只消耗 500 Token但第 10 轮可能消耗 4000 Token因为前面 9 轮的工具结果都在输入里。要控制成本除了上下文压缩还需要控制循环深度并对单轮任务设置合理的 max_iterations。显存占用方面如果使用本地模型主要取决于模型服务进程。Agent 架构本身不会显著增加显存占用因为 Agent 服务只是做 HTTP 转发和消息拼接。但并发批量任务时如果每个任务都新建连接请求本地模型服务模型服务的显存会被推理并发度影响。建议批量任务并发数先设为 1 或 2观察模型服务响应时间再逐步调高。资源观察的基本方法是加计时日志。具体建议在每次模型调用前后记录耗时和 Token 使用量。在每次工具调用前后记录耗时和结果大小。在 AgentLoop.run 结束前汇总输出本轮循环的总耗时、总 Token、工具调用次数。批量任务中记录每个任务耗时用于识别慢任务。如果发现慢任务集中在工具调用阶段优先排查工具本身的性能如果慢任务分布在模型调用阶段优先考虑降低循环轮次、压缩上下文或升级模型服务配置。10. 多智能体扩展思路单个 Agent Loop 能解决一部分问题但业务复杂度上来后单一 Agent 会面临一个矛盾既要具备专业领域深度又要处理跨度很大的任务。这种情况下多智能体协作是更合理的架构方向。类 PI-Agent 的架构可以很自然地扩展为多智能体协作。最常见的分工模式包括三种角色Planner 负责把复杂任务拆解为子任务规划执行顺序然后把每个子任务分发给对应的执行智能体。Executor 负责执行具体子任务它内部可以是另一个 Agent Loop也可以只是一个工具调用。Reflector 负责检查执行结果判断是否达到预期如果没达到就反馈给 Planner 重新规划。这种设计的价值在于职责隔离和上下文隔离。每个子 Agent 只维护与自己任务相关的上下文不会被无关的历史消息干扰。相比单一超长 Agent Loop多智能体协作在复杂任务上往往更稳定也更容易调试和测试。多智能体之间的通信方式通常有两种。一种是消息传递Agent 之间通过任务结果队列或消息总线传递结构化数据另一种是共享黑板模式多个 Agent 读写共享的工作区状态。原型阶段建议先用消息传递结构清晰出了问题容易追踪。生产环境如果需要多个智能体并发处理子任务再引入消息队列如 Redis Stream 或 RabbitMQ。从单次调用到单 Agent Loop再到多 Agent 协作这条演进路径本质上是在不断增加系统的“决策粒度”。每次增加决策点系统的灵活性都会上升但随之而来的是成本、延迟和调试难度的同步上升。架构设计的目标不是把系统做到最复杂而是找到业务收益和技术成本之间的平衡点。11. 常见问题与排查方法问题现象可能原因排查方式解决方案接口报 model 不存在模型名配置错误查看模型服务启动日志确认可用的模型列表修改 model 参数为正确的模型名模型不触发工具调用模型不支持 Function Calling或工具 Schema 不合法打印 tools 参数检查 Schema 格式换一个已知支持工具调用的大模型测试切换兼容模型或在 Prompt 中要求固定 JSON 输出工具参数解析失败模型生成的参数不符合 Schema打印 tool_call.function.arguments比对 Schema增加 pydantic 参数校验降低 temperatureLoop 陷入重复调用工具返回结果不满足模型预期模型反复尝试同一工具查看日志中同一工具调用次数增加重复调用检测连续 3 次相同则终止并总结上下文超限错误消息列表超过模型最大上下文长度查看报错信息的 token 数使用 compress_messages 压缩限制 max_iterations接口响应超时Agent Loop 循环次数过多或模型推理慢查看耗时日志定位高耗时阶段降低 max_iterations改用异步任务接口批量任务部分失败模型服务限流或工具偶发异常查看失败任务日志区分错误类型增加重试和退避降低并发数工具执行后模型回答不准确工具结果回填格式不对查看 tool 角色消息内容确保 content 是 JSON 字符串且包含关键结果显存不足本地模型服务并发推理过高查看模型服务日志和 nvidia-smi降低并发数或换更小的模型排查的基本原则是先看日志再看消息内容最后才怀疑模型能力。Agent Loop 的每一步决策、每次工具调用都应该有日志没有日志时排查会非常困难。12. 最佳实践与合规边界架构层面最值得记住的几个实践如下。第一第一次跑通永远用小参数。max_iterations 设成 3工具只注册一个模型用最快的版本。先确认 Loop 链路通再逐步增加工具和任务难度。第二维护一套最小可运行配置包括最小依赖、最小模型名、最小工具集。这样后续升级或定位问题时永远有一个可以回退的基线。第三工具调用必须可观测。每个工具的执行时间、输入参数、输出结果都要记录不然 Loop 出了一次诡异的结果你根本无法定位是从哪一步开始错的。第四批量任务必须加失败重试和结果落盘。没有重试的批量任务是脆弱的没有落盘的结果是不可审计的。第五接口服务要限制访问范围尤其在局域网或公网开放 Agent 服务时。Agent 具备工具调用能力如果被未授权的人调用等于把你的内部工具暴露给外部。必须增加身份认证和操作审批。合规层面Agent 涉及的工具调用如果作用于真实业务系统需要特别注意权限设计。建议把工具的权限收敛到最小可用范围例如数据库账号只读、外部接口只限定指定操作。涉及人脸、声音、版权素材、个人隐私数据的处理场景必须确认素材来源合法、用户已授权处理过程要有审计日志。涉及自动化操作例如自动发消息、自动下单、自动修改配置的必须设置人工审批节点不能让模型在循环中直接执行高风险操作。发布或商用前还要对 Agent 的输出做抽样复核确认不会生成违规或误导性内容。13. 总结整篇文章的核心就一句话模型是推理引擎智能体是围绕模型构建的循环系统。从单次调用到 Agent Loop变化的不是模型能力而是架构设计。你要在模型外面补上工具注册、上下文管理、循环控制、终止条件、批量调度、日志审计这些工程组件才能把一个“会聊天的模型”变成一个“能干活的任务执行系统”。如果看完这篇文章只打算做一件事建议先实现一个最简 Agent Loop一个工具、三轮回合、输出日志。先把循环和工具回填跑通你会对模型决策和工具执行之间的关系有非常直观的感受。在这个过程中最容易踩的坑是上下文管理没做就疯狂加工具导致消息列表爆炸后续所有轮次变慢变贵。正确顺序是先把短循环跑稳再加工具再加上下文压缩最后再考虑多智能体扩展。后续如果你想继续深入可以往这几个方向扩展一是把上下文压缩模块换成向量数据库长期记忆二是把线程池批处理换成 RabbitMQ 或 Redis Stream 任务队列三是增加人工审批组件把高风险工具操作纳入审批流四是尝试 Planner-Executor-Reflector 多智能体协作把本文的 Loop 作为 Executor 的内核。每一步扩展都以“可观测、可回退、可审计”为前提这套类 PI-Agent 架构就能从原型平滑长成生产级系统。