1. 为什么我想搭一个 Agent 平台而不是再写一个 Agent去年下半年开始我陆续帮几个团队做过 AI Agent 的落地项目。做完第三轮之后我意识到一个很尴尬的事实每次新需求进来我们都在重复造轮子。工具调用的注册逻辑重写一遍记忆模块重写一遍Prompt 模板管理重写一遍连前端那个对话界面都要重新搭。一个 Agent 从立项到能跑通业务闭环少说两周多则一个月其中真正跟业务相关的部分可能只占三成。这就是我想做 Agent 平台的起点。不是想做一个“更聪明的 Agent”而是想做一个“造 Agent 的工厂”。当 Agent 有了工厂团队里每个人都能像搭积木一样组装出自己的数字同事——运营可以造一个自动整理竞品动态的助手客服可以造一个能查订单能改地址的助手研发可以造一个能读日志能提 Issue 的助手。他们不需要懂 FastAPI 的路由怎么注册不需要懂 React 的状态怎么管理只需要描述清楚“这个同事要干什么”。这篇文章我会把整个平台的搭建过程拆开讲。技术栈选的是Python FastAPI React这是目前做 AI Agent 平台最顺手的一套组合Python 侧生态最全FastAPI 写异步接口干净利落React 做配置界面灵活度高。我会讲清楚每个模块为什么这么设计、关键参数怎么定、踩过哪些坑以及一套可以直接抄的目录结构和核心代码骨架。适合已经了解 LLM 基本概念、想往 Agent 工程化方向走的开发者也适合正在评估“自建还是买现成”的技术负责人。先明确一个概念边界因为热词里很多人问“AI Agent 和 LLM 有什么区别”。LLM 是大脑Agent 是给大脑装上手脚和记忆之后的完整个体。DeepSeek 这类属于 LLM是模型层Agent 是在模型之上加了规划、工具调用、记忆、执行循环的系统。而 Agent 平台是在 Agent 之上再加一层抽象把 Agent 的构建过程标准化、可视化、可复用。三层关系理清了后面的设计才不会乱。2. 平台整体架构把 Agent 拆成可组装的零件2.1 核心设计思路配置驱动而非代码驱动我见过不少团队做 Agent 平台第一反应是做一个“代码生成器”——用户填表单平台生成一段 Python 代码让用户去跑。这个思路我试过问题很大。生成的代码用户看不懂、改不动一旦业务变了就得重新生成平台变成了一个一次性的脚手架。我最后选的是配置驱动路线。一个 Agent 在平台里就是一条数据库记录包含它的角色定义、可用工具列表、记忆策略、模型参数。运行时由平台的执行引擎读取这条配置动态组装出一个可运行的 Agent 实例。用户改配置Agent 行为立刻变不需要重新部署。这个选择背后的逻辑是Agent 的本质是“LLM 工具 记忆 循环控制”这四样东西都是可以用结构化数据描述的。角色定义是文本工具列表是 ID 数组记忆策略是枚举值模型参数是键值对。既然都能描述就没必要生成代码。提示配置驱动的前提是把 Agent 的运行时抽象做干净。如果执行引擎里到处是 if-else 判断“如果是这种 Agent 就走这条路”那配置驱动会变成维护噩梦。我的做法是定义统一的 Agent 执行接口所有差异都通过配置项注入。2.2 四层架构拆解整个平台我分成四层从下往上依次是层级职责关键技术模型接入层统一封装不同 LLM 的调用适配器模式、流式响应Agent 运行时层执行循环、工具调度、记忆读写异步任务、状态机平台服务层Agent 的增删改查、会话管理、工具注册FastAPI、SQLAlchemy交互层配置界面、调试对话、运行监控React、SSE模型接入层单独抽出来是因为企业里往往不止用一个模型。有的场景用 DeepSeek 性价比高有的场景需要更强的推理能力。适配器模式让上层完全不感知底层用的是哪个模型只认统一的chat和chat_stream接口。Agent 运行时层是平台的心脏。它要处理的是一个循环把当前对话历史和系统提示发给模型模型返回要么是最终回答要么是工具调用请求如果是工具调用就执行工具、把结果塞回历史、再发给模型直到模型给出最终回答或达到最大轮次。这个循环看着简单但异步、超时、错误重试、流式输出全都要处理好。2.3 为什么 FastAPI 是这个场景的最优解选 FastAPI 不是跟风。Agent 平台的接口有两个特点一是大量 IO 等待等模型返回、等工具执行二是需要流式推送对话要一个字一个字往外蹦。这两个特点决定了同步框架会很吃力。FastAPI 原生支持 async/await一个工作进程能扛住大量并发等待不用开一堆线程。它的StreamingResponse配合 SSE 做流式输出非常自然。再加上 Pydantic 做请求校验Agent 配置这种结构复杂的对象校验起来很省心。SQLAlchemy 2.0 的异步支持也成熟了数据库这块不会成为瓶颈。对比一下 FlaskFlask 做同步接口没问题但流式输出和并发等待要额外折腾。对比 DjangoDjango 太重Agent 平台不需要那么多内置功能反而被它的 ORM 和中间件束缚。FastAPI 的轻量和异步特性刚好卡在这个场景的甜点上。2.4 项目目录结构目录结构我调整过三版最后定下来是这样agent-platform/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置管理 │ ├── database.py # 数据库连接 │ ├── models/ # SQLAlchemy 模型 │ │ ├── agent.py │ │ ├── session.py │ │ └── tool.py │ ├── schemas/ # Pydantic 校验模型 │ ├── api/ # 路由层 │ │ ├── agents.py │ │ ├── sessions.py │ │ └── tools.py │ ├── core/ # 核心逻辑 │ │ ├── runtime.py # Agent 执行引擎 │ │ ├── memory.py # 记忆管理 │ │ └── llm_adapter.py # 模型适配器 │ ├── tools/ # 内置工具 │ │ ├── registry.py │ │ └── builtin/ │ └── services/ # 业务服务层 ├── frontend/ # React 前端 ├── alembic/ # 数据库迁移 └── tests/这个结构的关键是把core核心逻辑和api接口层严格分开。核心逻辑不依赖 FastAPI可以单独测试也可以被其他入口调用。tools目录独立出来是因为工具会越来越多需要一套注册机制来管理。3. Agent 运行时执行循环怎么写才稳3.1 执行循环的核心状态机Agent 执行循环我用状态机来管理状态流转是这样的IDLE - THINKING - TOOL_CALLING - THINKING - ... - DONE。每次进入 THINKING 就是把当前消息历史发给模型拿到响应后判断如果响应里有工具调用请求就进 TOOL_CALLING如果没有就是最终回答进 DONE。为什么要用状态机而不是简单的 while 循环因为实际运行中会出现很多边界情况模型连续调用同一个工具、工具执行超时、达到最大轮次还没结束、用户中途取消。用状态机把这些情况显式建模代码可读性和可维护性都好很多。class AgentState(Enum): IDLE idle THINKING thinking TOOL_CALLING tool_calling DONE done ERROR error async def run_agent(agent_config, messages, max_turns10): state AgentState.THINKING turn 0 while state not in (AgentState.DONE, AgentState.ERROR): if turn max_turns: state AgentState.ERROR break if state AgentState.THINKING: response await llm_adapter.chat( messages, toolsagent_config.tools ) if response.tool_calls: state AgentState.TOOL_CALLING else: state AgentState.DONE elif state AgentState.TOOL_CALLING: results await execute_tools(response.tool_calls) messages.extend(results) state AgentState.THINKING turn 1 return messagesmax_turns这个参数很关键。我一开始设的是 20结果发现有些 Agent 会陷入“调用工具-结果不满意-再调用”的死循环20 轮下来烧了不少 token。后来改成默认 10并且在前端暴露出来让用户按需调整。对于大多数业务场景10 轮足够完成一次任务。3.2 工具注册机制让工具即插即用工具是 Agent 的手脚平台必须让工具注册变得极其简单。我的设计是每个工具就是一个 Python 函数加上一个装饰器声明它的元信息。register_tool( namequery_order, description根据订单号查询订单状态, parameters{ order_id: {type: string, description: 订单号} } ) async def query_order(order_id: str) - dict: # 实际查询逻辑 return {status: shipped, eta: 2024-01-15}装饰器做的事情是把函数的名称、描述、参数 schema 注册到全局的工具注册表里。Agent 配置时只需要勾选工具名运行时执行引擎根据工具名从注册表里找到函数并调用。这里有个细节要注意工具的description直接决定了模型能不能正确使用这个工具。我踩过的坑是描述写得太简略比如只写“查询订单”模型不知道参数该传什么格式经常传错。后来我把描述改成“根据订单号查询订单状态订单号格式为纯数字”调用准确率明显提升。工具描述要当成给模型看的 API 文档来写。3.3 记忆管理短期记忆和长期记忆分开处理记忆这块我分成两层。短期记忆就是当前会话的消息历史存在内存或 Redis 里会话结束就清掉。长期记忆是跨会话的知识存在向量数据库里需要时检索出来注入到上下文。短期记忆的处理有个容易忽略的点消息历史不能无限增长。模型有上下文窗口限制历史太长要么报错要么被截断。我的做法是设置一个 token 阈值超过阈值就把最早的消息做摘要压缩。摘要用一个便宜的模型来做成本可控。长期记忆我用的是“写入-检索”模式。Agent 在对话中如果判断某条信息值得记住比如用户的偏好、重要的业务规则就调用一个save_memory工具把它写入向量库。下次对话时根据当前问题检索相关记忆注入到系统提示里。这套机制让 Agent 有了“越用越懂你”的能力。注意长期记忆的写入要克制。我见过有的实现把每轮对话都写进去结果向量库迅速膨胀检索出来的全是噪音。我的经验是只写入明确的结构化信息比如“用户偏好用表格展示数据”这种而不是整段对话。3.4 流式输出SSE 的正确打开方式对话界面必须流式输出否则用户等十几秒才看到回复体验很差。FastAPI 做 SSE 很直接用StreamingResponse包一个异步生成器就行。from fastapi.responses import StreamingResponse router.post(/sessions/{session_id}/chat) async def chat(session_id: str, message: str): async def event_generator(): async for chunk in runtime.run_stream(session_id, message): yield fdata: {json.dumps(chunk)}\n\n return StreamingResponse( event_generator(), media_typetext/event-stream )这里有个坑SSE 连接容易被中间的代理或负载均衡断开。我的做法是在生成器里定期发送心跳注释: heartbeat\n\n保持连接活跃。另外前端要用EventSource或 fetch 的流式读取来处理React 里我封装了一个自定义 hook 来管理连接状态和重连。4. 平台服务层FastAPI 接口与数据模型设计4.1 数据模型三张核心表撑起整个平台平台的数据模型不复杂核心就三张表agentsAgent 配置、sessions会话、messages消息。工具因为大部分是代码里注册的数据库里只存一个启用状态和配置覆盖。agents表的关键字段包括id、name、description、system_prompt、model_name、temperature、max_turns、tool_idsJSON 数组、memory_configJSON、created_by、created_at。其中tool_ids和memory_config用 JSON 字段存因为它们的结构会随平台演进变化用 JSON 比频繁改表结构灵活。messages表我加了一个metadataJSON 字段用来存工具调用的中间结果。这样调试的时候能看到 Agent 每一步做了什么对排查问题帮助很大。4.2 接口设计RESTful 为主特殊场景单独处理接口设计遵循 RESTful 风格Agent 的增删改查对应标准的 GET/POST/PUT/DELETE。会话和消息也类似。唯一特殊的是对话接口因为要流式输出用 POST 加 SSE 返回。# agents.py 路由示例 router.post(/agents, response_modelAgentOut) async def create_agent(agent: AgentCreate, db: AsyncSession Depends(get_db)): db_agent Agent(**agent.model_dump()) db.add(db_agent) await db.commit() await db.refresh(db_agent) return db_agent router.get(/agents, response_modellist[AgentOut]) async def list_agents(skip: int 0, limit: int 20, db: AsyncSession Depends(get_db)): result await db.execute(select(Agent).offset(skip).limit(limit)) return result.scalars().all()分页参数skip和limit是标配但要注意默认值别设太大。我一开始limit默认 100结果 Agent 多了之后列表接口返回慢。改成 20 之后流畅很多前端做无限滚动加载。4.3 异步数据库SQLAlchemy 2.0 的配置要点SQLAlchemy 2.0 的异步用法和 1.x 差别不小配置的时候有几个点容易出错。引擎要用create_async_enginesession 要用async_sessionmaker查询要用await db.execute(select(...))而不是db.query(...)。from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker engine create_async_engine( postgresqlasyncpg://user:passlocalhost/agent_platform, echoFalse, pool_size10, max_overflow20, pool_pre_pingTrue ) AsyncSessionLocal async_sessionmaker(engine, expire_on_commitFalse)pool_pre_pingTrue这个参数建议加上它会在从连接池取连接时先 ping 一下避免拿到已经断开的连接。expire_on_commitFalse也很重要否则 commit 之后对象属性会过期再访问会触发额外的查询。4.4 配置管理环境变量加 Pydantic Settings配置项包括数据库连接、模型 API 密钥、Redis 地址等全部通过环境变量注入用 Pydantic 的BaseSettings来管理。from pydantic_settings import BaseSettings class Settings(BaseSettings): database_url: str redis_url: str redis://localhost:6379 default_model: str deepseek-chat max_context_tokens: int 8000 class Config: env_file .env settings Settings()这样做的好处是本地开发用.env文件生产环境用容器注入的环境变量代码完全不用改。密钥这类敏感信息绝对不能硬编码在代码里这是基本的安全底线。5. React 前端配置界面与调试对话的实现5.1 页面结构三个核心页面前端我做了三个核心页面Agent 列表页、Agent 配置页、调试对话页。列表页展示所有 Agent 和它们的运行状态配置页是表单用来编辑 Agent 的各项参数调试对话页是跟 Agent 实时对话的界面同时展示工具调用的过程。技术选型上我用 Vite 做构建工具React Router 做路由状态管理用 Zustand比 Redux 轻很多这个场景够用UI 组件用 Ant Design表单组件丰富省去大量样式工作。5.2 配置表单动态表单的处理Agent 配置表单里最麻烦的是工具选择部分。工具列表是从后端动态获取的每个工具有不同的参数用户勾选工具后可能需要配置参数覆盖。我用一个受控组件来管理工具列表变化时重新渲染。function ToolSelector({ tools, selected, onChange }) { return ( div classNametool-grid {tools.map(tool ( div key{tool.name} classNametool-card Checkbox checked{selected.includes(tool.name)} onChange{e { const next e.target.checked ? [...selected, tool.name] : selected.filter(t t ! tool.name); onChange(next); }} {tool.display_name} /Checkbox p classNametool-desc{tool.description}/p /div ))} /div ); }这里有个体验细节工具描述要完整展示因为用户需要知道这个工具能干什么才能决定要不要勾选。我一开始把描述截断了结果用户反馈不知道工具具体功能后来改成完整展示加悬浮提示。5.3 流式对话EventSource 的封装前端接收 SSE 流我用 fetch 的ReadableStream而不是EventSource因为EventSource只支持 GET 请求而对话接口需要 POST 传消息体。async function streamChat(sessionId, message, onChunk) { const response await fetch(/api/sessions/${sessionId}/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ message }) }); const reader response.body.getReader(); const decoder new TextDecoder(); while (true) { const { done, value } await reader.read(); if (done) break; const text decoder.decode(value); // 解析 SSE 格式提取 data 行 text.split(\n).forEach(line { if (line.startsWith(data: )) { onChunk(JSON.parse(line.slice(6))); } }); } }这个封装要注意处理粘包问题一次read()可能拿到多条消息也可能拿到半条。我的做法是维护一个缓冲区按\n\n分割最后一段不完整的留在缓冲区里等下次拼接。5.4 工具调用可视化让 Agent 的思考过程可见调试对话页我加了一个工具调用展示区Agent 每次调用工具都会在这里显示调用了什么工具、传了什么参数、返回了什么结果。这个功能对调试极其重要没有它你根本不知道 Agent 为什么给出某个回答。实现上后端在流式输出时把工具调用事件也推给前端前端根据事件类型渲染不同的卡片。工具调用卡片默认折叠点击展开看详情。这样界面不会太乱需要时又能看到细节。提示工具调用可视化不仅是调试工具也是给非技术用户看的“信任建立”工具。运营同学看到 Agent 确实去查了数据库、确实调用了接口才会信任它的回答。6. 实操踩坑记录与常见问题排查6.1 模型返回格式不稳定怎么办这是最常见的问题。你要求模型返回 JSON它有时候返回带 markdown 代码块的 JSON有时候在 JSON 前后加解释文字。我的处理是三层防御第一层在 Prompt 里明确要求“只返回 JSON不要任何其他内容”第二层用正则提取 JSON 部分第三层解析失败时重试一次重试时把错误信息也发给模型让它修正。def extract_json(text: str) - dict: # 去掉 markdown 代码块标记 text re.sub(rjson\s*|\s*, , text) # 尝试直接解析 try: return json.loads(text) except json.JSONDecodeError: # 尝试提取第一个完整的 JSON 对象 match re.search(r\{.*\}, text, re.DOTALL) if match: return json.loads(match.group()) raise6.2 工具执行超时怎么处理外部工具调用比如查数据库、调第三方接口可能很慢甚至卡死。我给每个工具执行加了超时控制默认 30 秒超时就返回一个错误结果给模型让模型决定是重试还是换方案。async def execute_tool_with_timeout(tool_func, args, timeout30): try: return await asyncio.wait_for(tool_func(**args), timeouttimeout) except asyncio.TimeoutError: return {error: f工具执行超时{timeout}秒}超时时间要根据工具类型调整。查本地数据库 5 秒够了调外部 API 可能要 30 秒。我在工具注册时允许声明timeout参数不声明就用默认值。6.3 常见问题速查表问题现象可能原因排查方向Agent 不调用工具工具描述不清、系统提示没引导检查工具 description在 system prompt 里明确工具使用场景调用工具但参数错误参数 schema 定义不准确检查 parameters 定义补充格式说明和示例回复重复或循环max_turns 太大、工具返回结果无变化降低 max_turns检查工具是否返回了有效信息流式输出中断代理超时、心跳缺失加心跳检查 nginx 的 proxy_read_timeout上下文超限报错消息历史太长开启摘要压缩降低 max_context_tokens并发高时响应慢数据库连接池不够、模型调用排队调大 pool_size模型调用加限流和队列6.4 几个我踩过的坑第一个坑是异步函数里混用同步代码。我在工具函数里直接用了同步的数据库查询结果整个事件循环被阻塞并发能力直接归零。后来把所有 IO 操作都改成异步性能才上来。FastAPI 的异步优势建立在“全链路异步”上有一环是同步的就会拖累整体。第二个坑是Redis 连接没做连接池。短期记忆存 Redis一开始每次操作都新建连接高并发下连接数暴涨。后来改用redis.asyncio的连接池问题解决。第三个坑是前端状态更新导致重复渲染。流式输出时每个 chunk 都触发一次 setState消息长了之后页面卡顿。后来用useRef累积内容用requestAnimationFrame节流更新流畅度明显改善。7. 平台后续可以怎么扩展平台跑通之后我陆续加了一些扩展能力这里分享几个我觉得价值最高的方向。Agent 模板市场。把常用的 Agent 配置做成模板新用户一键复制就能用。比如“竞品监控助手”“客服工单分类助手”“代码审查助手”每个模板预置好系统提示和工具组合。这个功能让非技术用户的上手成本几乎降到零。多 Agent 协作。单个 Agent 能力有限但多个 Agent 可以分工。我实现了一个简单的编排机制一个“协调者”Agent 负责拆解任务把子任务分给不同的“执行者”Agent最后汇总结果。这套机制用在复杂业务流程上效果很好比如“先查数据、再分析、再生成报告”这种多步骤任务。运行监控与成本统计。每个 Agent 的调用次数、token 消耗、平均响应时间都记录下来做成仪表盘。这个功能对团队管理很重要能清楚看到哪个 Agent 用得多、哪个成本高需要优化。工具市场与自定义工具。除了内置工具允许用户用 Python 写自定义工具上传。我设计了一个沙箱执行环境用户上传的工具代码在受限环境里运行保证平台安全。这个功能让平台的扩展性上了一个台阶业务团队可以自己接入内部系统的接口。整套平台从零到能跑通业务我大概花了三周时间其中一半时间在调试执行循环和流式输出。现在回头看最值得的投入是把运行时抽象做干净了后面加工具、加模型、加协作机制都很顺。如果你也在做类似的事情我的建议是先把执行循环和工具注册这两块打磨好其他的都是锦上添花。
