我一直觉得AgentChat这种概念看起来高深其实拆开看就是一整套“让AI按规矩做事”的工程套路。最近这词儿热度确实高各家都在做多智能体协作但真正落到自己项目里很多人还是懵的到底怎么从一个普通的对话接口一步步变成能调用工具、能多角色分工、能自己决定下一步动作的Agent系统这篇我就用一套完整的实操过程带你从零手搓一个最小可运行的AgentChat不依赖重型框架代码量控制在几百行以内但该有的机制全都有。1. 先从设计上想清楚AgentChat到底在解决什么问题1.1 一句话解释AgentChat以及它和普通聊天机器人的区别如果你用过ChatGPT的网页版那其实你已经接触过最基础的“对话式AI”了。你问一句它答一句上下文由平台帮你管理。但AgentChat不一样它的核心不是“回答”而是“完成任务”。打个比方普通聊天机器人像是一个只动嘴的客服你说什么它都接话但它不会真的去帮你订机票、查库存、改文档。而AgentChat更像一个“有手有脚”的实习生它不但能听懂你的需求还能自己决定先干什么、后干什么需要的时候调工具、查数据、甚至把任务拆给其他几个“同事”一起干。所以从架构上看AgentChat通常具备几个关键能力多轮对话状态管理记住上下文而且是有结构地记工具调用能力让模型不只是输出文字还能输出结构化指令去触发外部API多Agent协作按任务类型分派给不同角色比如一个写代码、一个查资料、一个做校对自主决策循环模型根据当前结果决定是继续执行、换一种方式还是结束任务。1.2 我为什么选择从零手写而不是直接套LangChain或AutoGen市面上现成的多Agent框架确实不少LangChain、AutoGen、CrewAI、MetaGPT一个比一个响亮。但如果你只是为了搞懂原理、做一个自己能完全掌控的轻量服务我强烈建议先手写一版。原因有三第一框架封装的抽象层级太高出问题很难排查。你调一个AgentExecutor它内部到底怎么维护上下文、怎么解析工具返回、怎么处理模型输出异常对新人来说就是个黑盒。遇到bug你连日志都看不懂更别说调优了。第二真实业务里的Agent Chat不一定需要那么重的编排。很多场景就是“一个路由器加几个工人”的结构你自己用一百行代码就能写得明明白白后续改逻辑也快。第三手写一遍能真正理解Agent的运作机制。等你把消息组装、工具注册、路由分发这套流程走通了再去看LangChain源码那感觉是完全不一样的——你不会再觉得那是魔法而是能一眼看出它每一步在做什么。当然如果是企业级生产项目、需要复杂记忆机制或稳定的多智能体调度那用成熟框架没问题。但学习路径上先手写再上框架是最扎实的走法。1.3 一个只用Python标准库也能跑的最小架构这里我先给出整个系统的核心架构后面所有代码都会围绕这个结构来写。用户输入 ↓ Router Agent路由智能体判断意图决定交给谁 ↓ Worker Agent执行智能体调用工具、处理任务 ↓ Tool Layer工具层天气查询、计算器、HTTP请求等 ↓ 结果汇总 → 返回用户在这个架构里Router和Worker都是基于同一个大模型API的封装区别只在于System Prompt不同、可用的工具不同。Router不直接操作工具只负责“派单”Worker负责“干活”。如果任务比较复杂Router还可以把任务拆解成多个子任务分发给多个Worker最后再合并结果。这个设计的好处是简单、可控、容易扩展。你以后想加一个新Agent只需要加一个Prompt定义和对应的工具集合Router那边加一个路由条件就行不需要动核心逻辑。2. 核心机制拆解让Agent真正“会思考、会分工”2.1 消息循环对话不再只是“一问一答”普通对话接口是“你发一条我回一条”Agent不一样它需要在一个任务内部进行多轮自我对话。比如用户说“帮我查一下北京和上海的天气然后告诉我哪个更适合户外跑步”这个任务在内部至少要经历这些步骤Agent先理解意图发现需要查两个城市的天气调用天气工具传参city北京拿到结果再调用天气工具传参city上海拿到结果对比分析两个城市的数据生成最终回答。这个过程在OpenAI的Function Calling机制下表现为“模型返回tool_calls → 程序执行工具 → 把工具结果作为消息追加到对话 → 再发给模型”。如此循环直到模型不再请求工具调用。这就是AgentChat的核心消息循环。我最初实现时踩过一个坑只把工具结果追加到消息列表却没有把“这是工具调用的返回结果”标注清楚结果模型后续对话混乱以为工具结果是用户说的。正确的做法是用role: tool的消息类型并且带上tool_call_id来关联具体的工具调用请求。2.2 角色定义用System Prompt给Agent“立人设”Agent和普通Prompt调用的最大区别在于每个Agent都有一个明确的“岗位说明书”也就是System Prompt。这个Prompt不只是告诉模型“你是谁”更重要的是设定行为边界。以我们的Router为例它的Prompt可以这样设计你是一个任务路由助手。用户的请求会发给你你需要判断该任务类型 并返回一个JSON对象格式如下 {agent: weather_worker, reason: 需要查询天气数据} 可选agent值 - weather_worker需要查询天气、气温、降水等气象信息 - calculator_worker需要数学计算、数值分析 - general_worker其他通用问题你看这个Prompt里做的事情就两件约束输出格式明确分类标准。实操中我发现System Prompt写得越具体模型路由的准确率越高。如果你只是写“你是一个智能助手请帮我分类”那模型大概率会自由发挥输出各种奇怪格式。另外一个小技巧给每个Agent设定“性格”和“禁忌”。比如给Worker加一句“如果你调用的工具返回错误不要编造数据必须如实告知用户工具调用失败”这个在真实场景里非常管用能大幅减少幻觉。2.3 路由策略多Agent如何决定谁回答路由是AgentChat区别于单Agent对话的关键。策略上我见过三类做法第一类是“硬编码规则路由”用正则或关键词匹配。比如消息里出现“天气”就转天气Agent出现“计算”就转计算器Agent。优点是快、零成本缺点是只能应对预设场景碰到“帮我看看明天下不下雨顺便算一下跑10公里要多久”这种混合意图就抓瞎。第二类是“模型路由”就是我们上面设计的Router Agent。它用LLM来理解用户意图灵活性高得多。代价是每次对话多一次模型调用有额外延迟和费用。第三类是“混合路由”先用规则匹配快速命中高频场景命中不了再交给模型路由兜底。我实际项目里推荐这种既省成本又保证兜底能力。如果你做的是垂直场景工具型Agent建议硬编码为主如果你做的是泛对话型Agent直接用模型路由就好省得自己维护词表。3. 实操落地三小时搓出一个可运行的AgentChat3.1 环境准备与依赖安装整个项目我推荐用Python 3.10以上版本依赖没几个核心就是FastAPI和OpenAI SDK。如果你用的是国内的大模型服务比如DeepSeek、智谱等只要接口兼容OpenAI格式代码几乎不用改换base_url就行。mkdir agentchat-demo cd agentchat-demo python -m venv venv source venv/bin/activate # Windows下用 venv\Scripts\activate pip install fastapi uvicorn openai pydantic注意这里我特意没装LangChain之类的重量级库。你需要的只是HTTP服务框架、OpenAI客户端和数据结构定义其他都自己写。3.2 Agent基类与OpenAI调用封装我们先写一个Agent基类所有具体Agent都继承它。这个基类负责维护对话历史、调用模型、处理返回值。from openai import OpenAI from typing import List, Dict, Any, Optional import json class BaseAgent: def __init__(self, name: str, system_prompt: str, model: str gpt-4o-mini): self.name name self.system_prompt system_prompt self.model model self.client OpenAI() self.messages: List[Dict[str, str]] [] self.messages.append({role: system, content: system_prompt}) def reset(self): 清空对话历史保留system prompt self.messages [{role: system, content: self.system_prompt}] def chat(self, user_input: str) - str: 单轮对话返回模型文本回复 self.messages.append({role: user, content: user_input}) response self.client.chat.completions.create( modelself.model, messagesself.messages, ) reply response.choices[0].message.content self.messages.append({role: assistant, content: reply}) return reply这里有几个设计细节值得说。第一消息历史存在Agent实例内部意味着每个Agent实例是有状态的这在处理多轮对话时很重要。如果你用全局变量存消息一旦遇到并发请求就会串号。第二reset()方法一定要提供。因为Agent实例一旦创建就常驻内存用户会话结束或者切换任务时必须能清空历史否则前面的对话会污染后面的判断。第三模型调用超时和重试逻辑这里没写但生产环境必须加。你可以用openai库自带的重试机制或者自己包一层tenacity。我就遇到过第三方大模型服务偶尔抽风不加重试的话整个Agent流程直接断掉。3.3 工具注册与Function Calling实现工具是Agent的“双手”。这里我用一个天气查询工具作为示例同时加上一个计算器工具。先定义工具列表。def get_weather(city: str) - str: 模拟天气查询接口。 实际项目中可以替换为真实天气API。 weather_data { 北京: {temp: 26, condition: 晴, humidity: 40}, 上海: {temp: 30, condition: 多云, humidity: 65}, 广州: {temp: 33, condition: 小雨, humidity: 80}, } data weather_data.get(city) if data: return json.dumps({city: city, **data}, ensure_asciiFalse) return json.dumps({error: f没有{city}的天气数据}, ensure_asciiFalse) def calculator(expression: str) - str: 安全计算器仅支持加减乘除和括号 import ast try: # 注意这里做了白名单校验绝对不要直接eval tree ast.parse(expression, modeeval) for node in ast.walk(tree): if not isinstance(node, (ast.Expression, ast.BinOp, ast.UnaryOp, ast.Constant, ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pair)): raise ValueError(不支持的表达式) result eval(expression, {__builtins__: {}}, {}) return json.dumps({result: result}, ensure_asciiFalse) except Exception as e: return json.dumps({error: str(e)}, ensure_asciiFalse)工具定义好之后需要把它们转换成OpenAI Function Calling需要的JSON Schema。这个格式有固定套路每个工具要声明名称、描述和参数结构。tools [ { type: function, function: { name: get_weather, description: 查询指定城市的实时天气信息, parameters: { type: object, properties: { city: {type: string, description: 城市名称如北京} }, required: [city] } } }, { type: function, function: { name: calculator, description: 执行数学计算支持加减乘除和括号, parameters: { type: object, properties: { expression: {type: string, description: 数学表达式如 (123)*4} }, required: [expression] } } } ] TOOL_FUNCTIONS { get_weather: get_weather, calculator: calculator, }这里我要重点提醒计算器工具绝对不能用Python的eval()直接执行用户输入否则就是一个致命的代码注入漏洞。我这里的写法用ast做了AST节点白名单校验只允许数字、加减乘除和括号其他节点类型直接拒绝。这个经验是从真实事故里学来的曾经有人在表达式里写__import__(os).system(rm -rf /)直接被eval执行了。3.4 带工具调用的Agent主循环有了工具定义接下来要写一个带工具调用能力的Worker Agent。它的核心是一个循环调用模型 → 检查是否有tool_calls → 有就执行工具、把结果追加进消息 → 继续调用模型 → 直到模型返回纯文本回复。class ToolAgent(BaseAgent): def __init__(self, name: str, system_prompt: str, tools: list, tool_functions: dict, model: str gpt-4o-mini): super().__init__(name, system_prompt, model) self.tools tools self.tool_functions tool_functions def chat(self, user_input: str, max_turns: int 5) - str: self.messages.append({role: user, content: user_input}) for _ in range(max_turns): response self.client.chat.completions.create( modelself.model, messagesself.messages, toolsself.tools, tool_choiceauto, ) message response.choices[0].message # 没有工具调用直接返回文本 if not message.tool_calls: self.messages.append({role: assistant, content: message.content}) return message.content # 有工具调用先把assistant消息加入历史 self.messages.append({ role: assistant, content: message.content, tool_calls: [ { id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments } } for tc in message.tool_calls ] }) # 逐个执行工具并把结果作为tool消息加入历史 for tc in message.tool_calls: func_name tc.function.name func_args json.loads(tc.function.arguments) print(f[{self.name}] 调用工具: {func_name}({func_args})) if func_name in self.tool_functions: result self.tool_functions[func_name](**func_args) else: result json.dumps({error: f未知工具: {func_name}}) self.messages.append({ role: tool, tool_call_id: tc.id, content: result }) # 循环继续让模型基于工具结果生成下一步 return 已达到最大工具调用轮数任务可能未完成请稍后再试。这个循环是整个AgentChat的心脏理解和稳住它是关键。我讲几点工程上很重要的细节。第一max_turns上限必须有。以前我写的时候没加结果模型在某个任务里反复调用工具连续调了20多次钱烧得飞快。加上上限之后异常情况能及时止损。第二assistant消息里的tool_calls字段必须原样回传。很多人容易漏掉这一点你从API返回的message里拿到tool_calls想当然地只把文本内容存下来工具结果追加进去后再发给模型模型就会报错说找不到对应的tool_call_id。正确做法是把整个tool_calls结构一并存进历史保证每次请求的消息序列结构完整。第三工具函数的异常必须捕获。我看到过很多示例代码是直接执行工具不做try/except的这不行。工具可能抛异常、可能返回错误数据这些都要包装成正常返回让模型根据反馈来决定下一步。否则整个对话链路就崩了。3.5 用FastAPI包一层HTTP服务命令行里跑通之后自然要把它包成HTTP服务供外部调用。FastAPI是首选代码量小、自动生成文档、支持并发。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() # 全局Worker实例此处按单会话演示 worker_prompt 你是一个多功能执行助手。你可以调用天气查询工具和计算器工具。 规则 1. 当用户询问天气时必须调用get_weather工具 2. 当用户需要计算时必须调用calculator工具 3. 工具返回后用简洁自然的语言向用户汇报结果 4. 如果工具返回错误请如实告知不要编造数据。 worker ToolAgent( nameworker, system_promptworker_prompt, toolstools, tool_functionsTOOL_FUNCTIONS, ) class ChatRequest(BaseModel): message: str class ChatResponse(BaseModel): reply: str app.post(/chat, response_modelChatResponse) async def chat(request: ChatRequest): reply worker.chat(request.message) return ChatResponse(replyreply) app.post(/reset) async def reset(): worker.reset() return {status: ok}这里给两个建议。第一生产环境不要用全局单例Worker因为所有用户会共享对话历史。你可以维护一个session_id → Agent实例的字典或者用Redis存储消息历史。我这里为了演示简单用了全局实例但你要清楚这是教学简化版。第二FastAPI的异步端点和同步阻塞调用之间要注意。worker.chat()是同步阻塞的如果直接放在async def端点里会阻塞事件循环。稳妥做法是用async def端点配合run_in_threadpool或者直接定义成def端点让FastAPI自己丢线程池处理。我写的示例是async def精简起见没加线程池实盘建议加上。3.6 跑起来看效果启动服务uvicorn main:app --reload --port 8000然后在另一个终端测试curl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d {message: 北京和上海的天气怎么样哪个适合跑步}服务端日志你会看到这样的输出[worker] 调用工具: get_weather({city: 北京}) [worker] 调用工具: get_weather({city: 上海})最终返回的回复大概是北京今天晴气温26°C湿度40%非常适合户外跑步。 上海今天多云气温30°C湿度65%体感稍微闷热一些。 综合考虑北京更适合跑步。这个效果就是我说的“普通聊天机器人做不到但AgentChat能做到”的典型例子模型自主决定连续调用了两次工具拿到结果后才组织语言回复。4. 经典翻车现场多Agent协作的5个深坑与解法4.1 Agent之间互相踢皮球陷入死循环多Agent协作最常见的坑就是Agent之间“你推我、我推你”迟迟不落库、不给最终结果。这在AutoGen那类群聊架构里特别常见。模型会生成类似“这个问题我转交给同事处理”的话然后下一个Agent又说“我无法处理请咨询另一个同事”循环往复。解决思路是引入“主持人Agent”或者“任务终结条件”。主持人负责每次对话后判断任务是否完成如果连续两轮没有任何Agent产出实质结果就直接强制收尾返回。手写方案里我在一个Agent决定向另一个Agent传递任务时给他加了一个计数器超过三次就强制让当前Agent给出最终答案。另一个实用做法是给每个Agent设定“不许中途转交”的硬性Prompt约束除非工具调用失败否则必须自己完成回答。这样能砍掉大量无效盘旋。4.2 上下文越聊越肥费用和延迟双双失控AgentChat的对话历史和工具结果都会累积起来每次全量发给模型。聊了几十轮之后传一次请求可能就是几千甚至上万token费用呈线性上涨响应速度也会明显变慢。我常用的处理策略是“滚动窗口摘要压缩”系统只保留最近N轮消息更早的消息压缩成一段摘要放进System Prompt。摘要可以在每次对话结束后由模型异步生成。你可以这样实现当消息数量超过阈值时调用一次模型把历史总结成200字以内的摘要然后清空旧消息只保留摘要和最近几轮消息。另外一个节约token的点是工具结果的详细内容不需要全量保留只保留对用户有用的结论。比如天气工具的原始JSON可以精简成“北京晴26°C”再存进历史。4.3 工具调用的参数格式不稳定OpenAI的Function Calling在大部分情况下会正确输出JSON格式的arguments但我遇到过不少次模型输出的参数里带多余的空格、换行甚至直接输出Markdown代码块包裹的JSON。如果你直接用json.loads()解析轻则报错重则整个Agent流程中断。我的做法是写一个容错解析函数def safe_parse_args(arguments: str) - dict: 容错解析模型返回的JSON参数 # 去掉可能的markdown代码块标记 text arguments.strip() if text.startswith(): text text.strip() if text.startswith(json): text text[4:] try: return json.loads(text) except json.JSONDecodeError: # 尝试提取第一个{...}块 start text.find({) end text.rfind(}) 1 if start ! -1 and end start: return json.loads(text[start:end]) raise这个函数虽然简单但能拦截掉很大比例的格式问题。另外一个治本的方法是在System Prompt里明确写“输出严格JSON格式不要包含任何Markdown标记或解释性文字”。4.4 并发场景下的状态混乱前面我提到过全局单例Worker的问题这里展开细说。如果你直接把Agent实例设置为全局变量两个用户同时发消息A的消息还没处理完B的消息就插进来Agent实例里的self.messages就乱套了。解决思路有几种最简单的是用session_id做维度每个会话维护独立的Agent实例放在内存字典里定期清理过期会话更健壮的是用Redis存消息历史每次请求都从Redis读取历史拼装消息列表Agent实例变成无状态的服务再进一步可以引入消息队列做异步处理把Agent调用丢进任务队列前端轮询结果。我个人的建议是中小型项目用内存字典加过期清理就够了别一上来就上Redis容易把简单问题复杂化。4.5 模型幻觉让Agent一本正经地胡说Agent在工具返回错误或者工具里没有用户想要的数据时很容易“编”。比如用户问一个不在天气数据列表里的城市模型可能自己编一个温度出来。这是真实场景里最危险的问题之一。我在Worker的Prompt里特意加了一条硬性规则如果工具返回的数据中没有用户查询的信息必须明确告知用户暂时没有该城市的数据 绝对不允许编造或推测。但光靠Prompt还不够最好在代码层面做校验。我通常的做法是工具返回的结构里带一个valid字段代码检查到validfalse时直接丢给模型一个强提示说“工具返回无效请如实告知用户”。双重保障比单靠Prompt可靠得多。5. 从“能跑”到“好用”AgentChat的进阶扩展5.1 给Agent装上长期记忆基础版的AgentChat在reset()之后什么都不记得这在很多真实场景下是不够的。比如用户三天前让你帮忙查过某家公司的资料今天又问“上次说的那家公司的融资情况有更新吗”Agent如果不记得就无法回答。长期记忆的经典实现是“向量数据库Embedding”把所有对话历史按片段切分Embedding后存入向量库。每次对话前先检索与当前问题相关的历史片段拼进Prompt作为上下文。如果你不想引入向量库也有轻量做法把每次对话的关键信息抽象成结构化记录存成JSON文件或者SQLite表。比如{user_id: 1, topic: 某公司融资, key_points: [A轮融资1000万, 焦点在AI芯片], timestamp: ...}。下次对话时根据用户ID拉取最近记录由模型筛选相关信息。小项目这么做够用了。5.2 从文本到多模态OpenAI最新的接口支持图片输入。你可以把AgentChat的消息类型扩展一下让用户不仅能发文字还能传图片、语音、文档。改动点主要在消息结构上OpenAPI格式的content字段可以变成数组里面放文本块和图片URL块。这个方向做出来效果提升非常明显。比如做个“看图查天气”的Agent用户发一张天空照片模型先识别图片判断天气情况再调用工具查询当地气象数据做交叉验证。这种“视觉工具”的组合普通单模型是做不到的。5.3 从单聊到群聊让多个Agent公开协作单AgentChat做到位之后下一步就是多Agent公开协作。所谓“公开协作”就是让用户能看到Agent们的讨论过程而不是只在幕后路由。架构上可以做个“圆桌会议”模式用户的请求作为议题多个Agent轮流发言各自发表观点或执行子任务一轮结束后用户再追加意见。这个形态很接近CrewAI里的“Process”模式或者AutoGen的Group Chat。手写实现时核心是一个GroupChatManager类它维护一个Agent列表和一个发言顺序策略。每到一轮Manager把之前所有人的发言历史拼装好让当前Agent基于完整上下文发言。为了防止某个Agent话痨占满全场可以给每条发言设置max_tokens限制并且由Manager决定发言者顺序而不是让模型自由选择。这个方向扩展空间很大但工程复杂度也上了一个台阶建议至少把单Agent版本跑通、稳定运行两周之后再动这个。最后再分享两个我在实际使用中体会较深的小技巧。第一工具返回格式一定要设计成“结果状态”的结构不要只返回一行裸文本。比如{valid: true, data: {...}}和{valid: false, error: ...}这样代码层判断逻辑会简单很多模型也不容易误解。第二记得给每个Agent加一个max_response_tokens限制。有些情况下模型会突然开始长篇大论把上下文塞爆限制单次输出长度能有效避免这个问题还能控制成本。踩过几次坑之后我现在写Agent的第一件事就是把各种上限都设好上限设计对了系统就成功了一半。