1. 为什么我要自己搭一个 Agent 平台而不是直接用现成产品2026 年这个时间点市面上能叫得出名字的 AI Agent 产品两只手数不过来。我身边不少朋友的第一反应是既然有现成的为什么还要自己搭这个问题我在动手之前也反复问过自己最后的结论很直接——现成产品解决的是通用场景下的通用任务而真正卡住日常工作的往往是那些带着强烈个人或团队上下文的琐碎流程。比如我每周要整理一批竞品更新日志、把散落在不同文档里的接口变更汇总成一份可执行清单、再根据历史工单自动生成回归测试点。这些事用通用 Agent 做要么上下文喂不进去要么每一步都要人工确认效率提升有限。自己搭平台的核心价值不在于造一个更聪明的模型而在于把模型能力编排成贴合自己工作流的固定同事。你可以把它理解成通用产品是租来的共享办公室什么都能干但什么都不顺手自建平台是你自己装修的工作间工具摆在哪、流程怎么走全由你定。这个定位一旦想清楚后面的技术选型就不会跑偏——我们不需要一个能回答一切问题的超级大脑我们需要一个能稳定执行特定任务链、能记住历史、能调用外部工具的数字同事。关键词里出现了 React、Next.js、Python、FastAPI 这几个技术栈这其实已经勾勒出一条非常典型的分层架构前端负责交互和可视化编排后端负责 Agent 的推理循环、工具调用和状态管理。我最终选的就是Next.js 做前端 FastAPI 做后端的组合。原因很实际Next.js 的 App Router 加上 Server Actions能把大量数据获取逻辑收拢到服务端前端只关心渲染FastAPI 的异步特性和 Pydantic 校验天然适合处理 Agent 这种一次请求触发多轮工具调用的长链路场景。两者之间用 SSE 或 WebSocket 做流式通信用户能实时看到 Agent 的思考过程而不是干等一个转圈。这篇文章我会把整个搭建过程拆开讲透包括目录结构怎么设计、Agent 的推理循环怎么写、工具怎么注册、流式输出怎么打通、以及我在实际跑起来之后踩到的那些坑。目标很明确你看完之后能照着搭出一个属于自己的、能真正干活的 Agent 平台而不是一个只能聊天的玩具。适合有基础 Python 和前端经验、想深入 Agent 开发但不知道从哪下手的同学如果你是完全零基础建议先把 Python 和 React 的基本语法过一遍再回来不然中间某些环节会有点吃力。2. 平台的整体骨架前后端职责怎么切分才不别扭2.1 为什么把 Agent 循环放在后端而不是前端很多人第一次做 Agent会下意识把推理循环写在前端因为前端能直接调模型 API看起来省事。我一开始也这么干过结果很快就撞墙了。Agent 的核心是一个思考—行动—观察的循环模型输出一个工具调用意图平台执行工具把结果塞回上下文再让模型继续思考直到任务完成。这个循环里涉及 API 密钥、工具执行权限、文件读写、数据库访问任何一项放在前端都是灾难——密钥暴露、权限失控、跨域问题一堆。所以正确的切分是后端持有模型密钥和工具执行能力前端只负责发起任务、展示过程和接收结果。后端用 FastAPI 暴露一个任务接口前端通过 SSE 订阅这个任务的执行流。这样职责清晰安全性也有保障。下面是我实际用的目录结构前后端分离但放在同一个仓库里方便统一管理agent-platform/ ├── backend/ │ ├── app/ │ │ ├── main.py # FastAPI 入口 │ │ ├── core/ │ │ │ ├── config.py # 配置与密钥管理 │ │ │ └── agent_loop.py # Agent 推理循环核心 │ │ ├── tools/ │ │ │ ├── registry.py # 工具注册中心 │ │ │ ├── web_search.py │ │ │ ├── file_ops.py │ │ │ └── code_runner.py │ │ ├── memory/ │ │ │ ├── short_term.py # 会话内短期记忆 │ │ │ └── long_term.py # 向量库长期记忆 │ │ ├── api/ │ │ │ ├── routes_task.py # 任务相关接口 │ │ │ └── routes_stream.py # SSE 流式接口 │ │ └── schemas/ │ │ └── task.py # Pydantic 模型 │ ├── requirements.txt │ └── .env ├── frontend/ │ ├── app/ │ │ ├── page.tsx # 任务面板首页 │ │ ├── agents/[id]/page.tsx # 单个 Agent 详情 │ │ └── api/ # Next.js 服务端路由 │ ├── components/ │ │ ├── AgentRunner.tsx # 任务执行组件 │ │ ├── ToolPanel.tsx # 工具配置面板 │ │ └── StreamViewer.tsx # 流式输出展示 │ ├── lib/ │ │ └── sse-client.ts # SSE 客户端封装 │ └── package.json └── docker-compose.yml这个结构里最关键的是agent_loop.py和registry.py两个文件。前者是整个平台的心脏后者决定了你的 Agent 能干什么。我见过太多项目把工具调用逻辑硬编码在循环里结果每加一个工具就要改核心代码维护成本极高。用注册中心模式之后新增工具只需要写一个函数加一个装饰器循环本身完全不用动。2.2 FastAPI 后端的启动配置与依赖管理后端我用的 Python 3.11这个版本对异步的支持已经非常成熟而且类型提示的体验比 3.9 好太多。依赖管理上我没有用 Poetry直接用 requirements.txt 加虚拟环境因为团队里有人不熟悉 Poetry统一用最朴素的方式反而沟通成本最低。核心依赖就几个fastapi0.115.0 uvicorn[standard]0.32.0 pydantic2.9.0 httpx0.27.0 sse-starlette2.1.3 openai1.52.0 numpy1.26.0这里重点说两个。sse-starlette是专门处理 SSE 的库比手写 StreamingResponse 省心很多它帮你处理了心跳、断线重连这些细节。httpx用来做异步 HTTP 请求工具里如果需要调外部 API用它比 requests 更合适因为不会阻塞事件循环。模型 SDK 我用的是 openai 的官方库因为现在大部分模型服务都兼容这套接口切换成本低。main.py里要做的事情不多主要是挂载路由、配置 CORS、初始化工具注册中心from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api import routes_task, routes_stream from app.tools import registry from app.core.config import settings app FastAPI(titleAgent Platform, version0.1.0) app.add_middleware( CORSMiddleware, allow_originssettings.ALLOWED_ORIGINS, allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 启动时自动发现并注册所有工具 registry.auto_discover(app.tools) app.include_router(routes_task.router, prefix/api/task, tags[task]) app.include_router(routes_stream.router, prefix/api/stream, tags[stream])auto_discover这个设计是我后来加的一开始是手动 import 每个工具模块工具一多就经常忘记注册。改成自动扫描目录之后只要文件放在tools/下并且用了注册装饰器启动时就会自动加载。这个改动看起来小但省了我不少为什么这个工具不生效的排查时间。2.3 Next.js 前端的路由与状态管理取舍前端我用的是 Next.js 14 的 App Router。选它而不是纯 Vite React主要看中的是服务端组件和 Server Actions 能简化数据流。Agent 平台的前端其实不复杂核心就三个页面Agent 列表、Agent 配置、任务执行。但任务执行页的状态管理是个难点因为要实时接收 SSE 流、更新消息列表、展示工具调用过程状态变化非常频繁。我没有引入 Redux 或 Zustand而是用 React 自带的useReducer加一个自定义的 SSE hook。原因是这个场景的状态更新模式很固定——都是收到一条事件往列表里追加或更新一条记录用 reducer 处理反而比全局状态库更清晰。下面是我封装的 SSE hook 的核心逻辑// lib/useAgentStream.ts import { useEffect, useReducer, useRef } from react; type StreamEvent | { type: thinking; content: string } | { type: tool_call; tool: string; args: unknown } | { type: tool_result; tool: string; result: string } | { type: final; content: string } | { type: error; message: string }; function reducer(state: StreamEvent[], event: StreamEvent) { return [...state, event]; } export function useAgentStream(taskId: string | null) { const [events, dispatch] useReducer(reducer, []); const sourceRef useRefEventSource | null(null); useEffect(() { if (!taskId) return; const source new EventSource(/api/stream/${taskId}); sourceRef.current source; source.onmessage (e) { const event JSON.parse(e.data) as StreamEvent; dispatch(event); if (event.type final || event.type error) { source.close(); } }; source.onerror () { dispatch({ type: error, message: 连接中断请重试 }); source.close(); }; return () source.close(); }, [taskId]); return events; }这里有个细节值得说EventSource在收到final或error事件后必须主动 close否则浏览器会自动重连导致任务被重复触发。我一开始没处理这个测试时发现任务跑完后又莫名其妙重新开始排查了半天才定位到是 SSE 自动重连的锅。这个坑在后面讲流式通信时还会展开。3. Agent 推理循环把思考—行动—观察写成可维护的代码3.1 循环的终止条件设计比循环本身更重要Agent 循环写起来不难难的是什么时候停。我见过的最常见错误是只判断模型有没有返回工具调用没有工具调用就结束。这个逻辑在简单任务上没问题但遇到复杂任务时模型可能连续调用十几次工具还没收敛或者陷入调用工具—结果不满意—再调用同一个工具的死循环。所以终止条件必须是多重的模型明确返回了最终答案没有工具调用意图达到最大迭代次数我设的是 15 次超过就强制中断并返回当前进展连续两次工具调用参数完全相同判定为死循环直接中断总耗时超过阈值默认 120 秒防止某个工具卡死拖垮整个任务这四条里第三条是我踩坑之后加的。当时有个任务让 Agent 去查一个不存在的文件它调用read_file失败然后重试再失败再重试整整跑了十几次才被最大迭代次数拦住。加上参数去重判断之后第二次相同调用就会被拦截直接告诉模型这个操作已经失败过请换一种方式。3.2 核心循环的完整实现与逐段拆解下面是我实际在用的循环实现我把它拆成几个部分讲# app/core/agent_loop.py import asyncio import json import time from typing import AsyncGenerator from app.tools.registry import registry from app.memory.short_term import ShortTermMemory from app.core.llm_client import llm_client MAX_ITERATIONS 15 MAX_DURATION 120 DUPLICATE_THRESHOLD 2 async def run_agent( task: str, agent_config: dict, memory: ShortTermMemory, ) - AsyncGenerator[dict, None]: start_time time.time() iteration 0 call_history: list[str] [] memory.add(user, task) tools_schema registry.get_schema(agent_config.get(allowed_tools)) while iteration MAX_ITERATIONS: if time.time() - start_time MAX_DURATION: yield {type: error, message: 任务超时已中断} return iteration 1 yield {type: thinking, content: f第 {iteration} 轮推理} response await llm_client.chat( messagesmemory.get_messages(), toolstools_schema, temperatureagent_config.get(temperature, 0.3), ) if not response.tool_calls: memory.add(assistant, response.content) yield {type: final, content: response.content} return for tool_call in response.tool_calls: signature f{tool_call.name}:{json.dumps(tool_call.args, sort_keysTrue)} call_history.append(signature) if call_history.count(signature) DUPLICATE_THRESHOLD: yield { type: error, message: f检测到重复调用 {tool_call.name}已中断, } return yield { type: tool_call, tool: tool_call.name, args: tool_call.args, } try: result await registry.execute(tool_call.name, tool_call.args) except Exception as e: result f工具执行失败: {str(e)} yield { type: tool_result, tool: tool_call.name, result: str(result)[:2000], } memory.add(tool, str(result), tool_nametool_call.name) yield {type: error, message: 达到最大迭代次数任务未完成}这段代码里有几个设计决策值得展开。第一我用AsyncGenerator而不是普通函数因为整个执行过程需要流式输出每产生一个事件就 yield 出去前端能实时看到。第二工具执行结果我截断到 2000 字符再塞回上下文因为有些工具比如读文件返回的内容可能非常长全塞进去会迅速撑爆上下文窗口而且大部分情况下模型只需要关键信息。第三工具执行失败我没有直接抛异常终止而是把错误信息作为观察结果返回给模型让它自己决定是重试还是换方案——这一点很重要Agent 的鲁棒性很大程度上来自于允许它犯错并自我纠正。3.3 短期记忆与上下文窗口的博弈短期记忆看起来简单就是维护一个消息列表但实际用起来问题不少。最典型的是上下文窗口溢出。一个跑了十几轮的任务每轮都有模型输出、工具调用、工具结果消息列表很快就变得很长。我的处理策略是分层压缩最近 5 轮的消息完整保留不做任何处理更早的工具调用结果只保留前 200 字符的摘要更早的模型思考内容如果超过 500 字符就截断系统提示词和用户原始任务永远完整保留这个策略不是拍脑袋定的是我观察了实际任务的消息增长曲线之后调的。大部分任务的关键信息集中在最近几轮早期的工具结果往往只是中间过程压缩掉不影响最终质量。实现上就是在get_messages()里做一次遍历和裁剪# app/memory/short_term.py class ShortTermMemory: def __init__(self, max_recent: int 5): self.messages: list[dict] [] self.max_recent max_recent def add(self, role: str, content: str, tool_name: str | None None): msg {role: role, content: content} if tool_name: msg[name] tool_name self.messages.append(msg) def get_messages(self) - list[dict]: if len(self.messages) self.max_recent * 2: return self.messages recent self.messages[-self.max_recent * 2:] older self.messages[:-self.max_recent * 2] compressed [] for msg in older: content msg[content] if msg[role] tool and len(content) 200: content content[:200] ...[已压缩] elif msg[role] assistant and len(content) 500: content content[:500] ...[已压缩] compressed.append({**msg, content: content}) return compressed recent注意压缩策略要根据你的实际任务调整。如果你的任务需要模型回溯很早之前的细节压缩得太狠会导致它失忆。我的建议是先跑一批真实任务观察哪些信息被引用得最多再决定压缩力度。4. 工具系统让 Agent 真正能动手的关键4.1 用装饰器注册工具新增能力不改核心代码工具系统的设计目标只有一个新增工具时除了写工具本身不需要动任何其他代码。我用的方案是装饰器加自动发现。每个工具就是一个普通的异步函数用tool装饰器标注名称、描述和参数 schema注册中心在启动时扫描整个tools目录把所有被装饰的函数收集起来。# app/tools/registry.py import importlib import inspect import pkgutil from typing import Callable class ToolRegistry: def __init__(self): self._tools: dict[str, dict] {} def tool(self, name: str, description: str, parameters: dict): def decorator(func: Callable): self._tools[name] { name: name, description: description, parameters: parameters, func: func, } return func return decorator def auto_discover(self, package: str): pkg importlib.import_module(package) for _, module_name, _ in pkgutil.iter_modules(pkg.__path__): importlib.import_module(f{package}.{module_name}) def get_schema(self, allowed: list[str] | None None) - list[dict]: tools self._tools.values() if allowed: tools [t for t in tools if t[name] in allowed] return [ { type: function, function: { name: t[name], description: t[description], parameters: t[parameters], }, } for t in tools ] async def execute(self, name: str, args: dict): if name not in self._tools: raise ValueError(f未知工具: {name}) func self._tools[name][func] if inspect.iscoroutinefunction(func): return await func(**args) return func(**args) registry ToolRegistry()get_schema里的allowed参数是给 Agent 配置用的。不同的 Agent 应该有不同的工具权限比如一个只负责查资料的 Agent 不应该有文件写入权限。这个设计让权限控制变得很简单配置里写个工具名列表就行。4.2 三个必备工具的实现细节与踩坑平台刚搭起来的时候我建议先实现三个工具网页搜索、文件读写、代码执行。这三个覆盖了大部分日常任务的需求。下面说几个实现时的关键点。网页搜索工具最容易踩的坑是返回内容太长。搜索引擎返回的原始结果往往包含大量导航、广告、无关链接直接塞给模型既浪费上下文又干扰判断。我的做法是在工具内部做一次清洗只提取标题、摘要和正文前 500 字# app/tools/web_search.py import httpx from app.tools.registry import registry registry.tool( nameweb_search, description搜索网页返回相关结果的标题和摘要, parameters{ type: object, properties: { query: {type: string, description: 搜索关键词}, limit: {type: integer, default: 5}, }, required: [query], }, ) async def web_search(query: str, limit: int 5) - str: async with httpx.AsyncClient(timeout15) as client: resp await client.get( https://api.example-search.com/search, params{q: query, limit: limit}, ) data resp.json() results [] for item in data.get(results, [])[:limit]: results.append( f标题: {item[title]}\n f摘要: {item.get(snippet, )[:500]}\n f链接: {item[url]} ) return \n---\n.join(results) if results else 未找到相关结果文件读写工具的关键是路径安全。绝对不能允许 Agent 读写任意路径否则一个配置错误就可能导致系统文件被改。我的做法是限定一个工作目录所有路径都相对于这个目录解析并且拒绝任何包含..的路径# app/tools/file_ops.py from pathlib import Path from app.tools.registry import registry from app.core.config import settings WORKSPACE Path(settings.WORKSPACE_DIR).resolve() def _safe_path(relative: str) - Path: target (WORKSPACE / relative).resolve() if not str(target).startswith(str(WORKSPACE)): raise ValueError(路径越界拒绝访问) return target registry.tool( nameread_file, description读取工作目录下的文件内容, parameters{ type: object, properties: {path: {type: string}}, required: [path], }, ) async def read_file(path: str) - str: target _safe_path(path) if not target.exists(): return f文件不存在: {path} content target.read_text(encodingutf-8) return content[:5000] if len(content) 5000 else content代码执行工具是最危险的我的建议是初期先不要做或者只在一个完全隔离的容器里跑并且限制执行时间。如果一定要做至少要做到超时强制终止、限制内存、禁止网络访问。这块展开能写一整篇这里先点到为止。4.3 工具描述怎么写模型才用得对工具能不能被正确调用很大程度上取决于描述写得好不好。我踩过的坑是描述写得太笼统模型经常在不需要的时候调用或者参数传错。后来我总结了几条经验描述里明确说清楚什么时候用而不只是这个工具是干什么的。比如当需要查询实时信息时使用比搜索网页有效得多。参数描述要给出具体示例。比如query参数描述写成搜索关键词例如2026 年 AI Agent 发展趋势模型传参的准确率明显提升。工具名用动词开头search_web比web_search更符合模型的调用直觉虽然差别不大但实测有影响。这些细节看起来琐碎但工具调用准确率从 70% 提到 90% 往往就靠这些。我做过对比测试同一批任务优化描述前后工具调用成功率差了将近 20 个百分点。5. 流式通信让用户看见 Agent 在想什么5.1 SSE 和 WebSocket 的选择依据流式通信有两个主流方案SSE 和 WebSocket。我最终选了 SSE理由很实际。Agent 任务的通信模式是单向的——后端不断推送事件前端只需要接收不需要在任务执行过程中往回发消息。SSE 天然就是为这种场景设计的基于 HTTP实现简单浏览器原生支持EventSource断线重连也是内置的。WebSocket 虽然双向但需要额外维护连接状态、处理心跳、写重连逻辑对这个场景来说是过度设计。当然 SSE 也有局限比如浏览器对同一域名的 SSE 连接数有限制HTTP/1.1 下是 6 个但我们的场景里一个用户同时跑的任务不会超过这个数所以不是问题。如果你的平台需要支持大量并发任务可以考虑用 HTTP/2连接数限制会宽松很多。5.2 后端 SSE 接口的实现与事件格式约定后端用sse-starlette实现流式接口核心是把 Agent 循环 yield 出来的事件转成 SSE 格式# app/api/routes_stream.py from fastapi import APIRouter, HTTPException from sse_starlette.sse import EventSourceResponse import json from app.core.agent_loop import run_agent from app.memory.short_term import ShortTermMemory from app.core.task_store import task_store router APIRouter() router.get(/{task_id}) async def stream_task(task_id: str): task task_store.get(task_id) if not task: raise HTTPException(status_code404, detail任务不存在) async def event_generator(): memory ShortTermMemory() try: async for event in run_agent( tasktask[input], agent_configtask[config], memorymemory, ): yield { event: message, data: json.dumps(event, ensure_asciiFalse), } except Exception as e: yield { event: message, data: json.dumps( {type: error, message: str(e)}, ensure_asciiFalse, ), } return EventSourceResponse(event_generator())事件格式我统一成{type, ...}的结构前端根据type字段决定怎么渲染。目前定义了五种事件类型thinking模型思考中、tool_call准备调用工具、tool_result工具返回结果、final最终答案、error出错。这个约定要前后端严格对齐不然前端会渲染出乱七八糟的东西。5.3 前端流式渲染的性能陷阱前端接收 SSE 事件后要实时更新 UI这里有个性能陷阱如果每个事件都触发一次完整的状态更新和重渲染事件一多页面就会卡。我一开始就是这么写的任务跑到第十几轮的时候页面明显掉帧。后来做了两个优化第一用useReducer批量合并更新。React 18 的自动批处理能合并同一事件循环内的多次 setState但 SSE 事件是分散在不同时间点到达的所以需要手动控制。我的做法是在 reducer 里只做数据追加把渲染压力交给 React 的 diff。第二工具结果的长文本做虚拟滚动或折叠。一个工具结果可能有几千字符全部渲染到 DOM 里很浪费。我默认只显示前 200 字符用户点击才展开全文。这个改动让长任务的页面流畅度提升非常明显。// components/StreamViewer.tsx function ToolResultItem({ result }: { result: string }) { const [expanded, setExpanded] useState(false); const preview result.slice(0, 200); const hasMore result.length 200; return ( div classNametool-result pre{expanded ? result : preview}/pre {hasMore ( button onClick{() setExpanded(!expanded)} {expanded ? 收起 : 展开全文} /button )} /div ); }提示SSE 的EventSource在连接断开后会自动重连如果你的任务已经结束但连接没关会导致任务被重复执行。务必在收到final或error事件后主动调用source.close()这一点我在前面提过但值得再强调一次因为它造成的 bug 非常隐蔽。6. 从能跑到好用我踩过的那些坑和优化经验6.1 模型输出格式不稳定导致的解析失败Agent 循环里最让人头疼的问题是模型有时候不按格式返回工具调用。虽然现在大部分模型都支持结构化的 function calling但在某些边界情况下比如上下文特别长、任务特别复杂模型可能会把工具调用写在普通文本里而不是走标准的 tool_calls 字段。我遇到过好几次模型输出了类似我将调用 search_web 工具参数是...这样的文本但tool_calls是空的循环就误判为任务完成直接返回了这段文本。我的解决方案是加一层兜底解析当tool_calls为空时检查模型输出里是否包含工具调用的特征比如工具名加 JSON 结构如果有就尝试提取并执行。这个兜底逻辑不能太激进否则会误判正常的最终答案。我的做法是只匹配非常明确的模式比如输出里同时出现了已注册的工具名和一对花括号包裹的 JSONimport re import json def try_extract_tool_call(content: str, known_tools: set[str]): for tool_name in known_tools: pattern rf{tool_name}\s*[\(]?\s*(\{{.*?\}}) match re.search(pattern, content, re.DOTALL) if match: try: args json.loads(match.group(1)) return tool_name, args except json.JSONDecodeError: continue return None这个兜底不是万能的但能救回不少本来会失败的任务。实测下来加了这层之后任务成功率大概提升了 5 到 8 个百分点。6.2 工具执行超时与并发控制工具执行必须设超时这一点我在前面提过但具体怎么设很有讲究。我一开始给所有工具设了统一的 30 秒超时结果发现网页搜索经常在 25 秒左右返回而代码执行有时候 5 秒就够了。统一超时要么太松要么太紧。后来改成每个工具自己声明超时时间注册的时候一起传进去registry.tool( nameweb_search, description..., parameters{...}, timeout20, ) async def web_search(query: str, limit: int 5) - str: ...执行的时候用asyncio.wait_for包一层async def execute(self, name: str, args: dict): tool self._tools[name] timeout tool.get(timeout, 30) return await asyncio.wait_for(tool[func](**args), timeouttimeout)并发控制是另一个容易被忽略的点。如果模型在一轮里同时调用了三个工具默认是串行执行的但其实它们之间没有依赖关系可以并发。我把同一轮的工具调用改成asyncio.gather并发执行整体任务耗时明显下降。但要注意如果工具之间有副作用比如都写同一个文件并发会出问题所以这个优化要谨慎使用最好让工具本身声明是否可并发。6.3 长期记忆的引入时机与向量库选型短期记忆只能维持单次任务跨任务的记忆需要长期记忆。但我的建议是不要一上来就做长期记忆先把单任务跑通跑稳再考虑跨任务。原因是长期记忆会引入检索、去重、更新等一系列复杂问题过早引入会让整个系统难以调试。等单任务稳定之后长期记忆的引入其实不复杂。核心就是把历史任务的关键信息任务描述、最终结果、用到的工具向量化存起来新任务开始时检索相似的历史记录作为参考。向量库我用的 Chroma因为它足够轻量本地跑不需要额外服务适合个人或小团队。如果数据量大了再考虑换 Milvus 或 Qdrant。# app/memory/long_term.py import chromadb from app.core.config import settings client chromadb.PersistentClient(pathsettings.CHROMA_DIR) collection client.get_or_create_collection(task_history) def remember(task: str, result: str, tools_used: list[str]): collection.add( documents[f任务: {task}\n结果: {result}], metadatas[{tools: ,.join(tools_used)}], ids[ftask_{hash(task)}], ) def recall(task: str, top_k: int 3) - list[str]: results collection.query(query_texts[task], n_resultstop_k) return results[documents][0] if results[documents] else []检索到的历史记录不要直接塞进上下文而是作为参考经验放在系统提示词里并且明确告诉模型以下是类似任务的历史处理方式仅供参考不一定适用于当前情况。这样能避免模型盲目照搬历史方案。6.4 一个真实任务的完整执行链路复盘最后用一个真实任务把整个链路串一遍。任务描述是帮我查一下最近一周 AI Agent 领域有哪些新发布的开源项目整理成一份清单每个项目附一句话说明。执行过程是这样的后端收到任务创建 task_id前端跳转到执行页并建立 SSE 连接。Agent 第一轮推理模型决定调用web_search参数是AI Agent 开源项目 2026。工具返回五条搜索结果事件推送到前端用户看到正在搜索。第二轮模型觉得信息不够又调了一次web_search参数更具体。第三轮模型基于两次搜索结果决定调用read_file读取一个本地维护的项目清单模板。第四轮模型整合所有信息输出最终清单事件类型是final前端关闭 SSE 连接。整个任务耗时约 18 秒调用了 3 次工具迭代 4 轮。这个链路里每个环节都可能出问题搜索超时、文件不存在、模型格式错误、SSE 断连。平台的价值就在于把这些异常都兜住让用户看到的是一个稳定的同事而不是一堆报错。我现在的做法是每个环节都有降级方案——搜索失败就返回暂时无法获取实时信息文件不存在就提示模型换方案格式错误就触发兜底解析。这些降级逻辑加起来可能占了整个代码库的三分之一但它们才是平台从能跑到好用的关键。如果你也在搭自己的 Agent 平台我的建议是先跑通最小闭环——一个工具、一个循环、一个流式接口然后再逐步加工具、加记忆、加优化。不要一开始就追求大而全那样很容易在细节里迷失最后什么都没跑起来。
