先说个反直觉的结论:一个真正能上线干活的AI Agent,代码量往往不超过500行,而这里面最值钱的部分不是调用大模型那两行,而是你围绕"循环、工具、记忆、兜底"这四个环节做的设计。我见过太多人把Agent想复杂了,一上来就堆重型框架,结果Demo跑了两天,连一个最简单的"查天气再决定带不带伞"都没闭环。这篇文章想给你一条可以照抄的路径,从零开始构建一个实用的AI Agent系统,并把从本地能跑到生产可用的每一步拆开讲清楚。不管你是什么基础,只要跟着把这几层逻辑搭起来,就能得到一个结构完整、能接真实数据、能部署上线的Agent脚手架。
这篇文章没有绑定任何特定的大模型厂商接口,代码里统一用chat_model这样的伪SDK来演示,你换成自己项目里实际使用的模型客户端,接口保持一致即可。这样做的目的是让思路不被某一家产品绑定,也不让品牌名称干扰你对本质的理解。
1. AI Agent相比普通程序到底多了什么——先看清边界再动手
1.1 传统程序和Agent的分界线,其实只有一条
很多新手对AI Agent的第一反应是"能自动聊天的机器人",这个理解不算错,但会把方向带偏。我习惯用一个更朴素的对比:普通程序是做选择题的,Agent是做填空题的。普通程序把所有的执行路径提前设定好,输入A走分支A,输入B走分支B,行为完全可预期;Agent拿到一个目标之后,需要自己判断"下一步该干什么",中途任何一步的结果都会改变它接下来的行动。
举一个生活化的例子。假设你要做一个"出行助手":
- 传统写法:用户选城市、选日期,程序调用天气接口,把结果格式化输出,结束。
- Agent写法:用户说"明天去青川市出差,要不要带伞",Agent先判断这需要青川市明天的天气,然后调用天气工具拿到降雨概率,发现上午有阵雨,自动补一句"建议带伞,另外下午转晴,行程不用改",中间每一步都是模型根据工具返回的实时结果现场决策的。
这里的关键差别在于:传统程序的决策逻辑写在代码里,Agent的决策逻辑写在模型的推理里。代码世界没有歧义和临场发挥,而Agent面对的是一句随口说的话里隐含的任务拆解。理解了这条分界线,你就能明白为什么Agent系统不是把API换成大模型接口就完事了,你需要给模型搭一个"决策-行动-观察"的舞台,这个舞台的骨架就是控制循环。
1.2 拆开一个最小的Agent,它只有四个部件
我习惯跟团队新人说,不管外面宣传的Agent概念多花哨,一个能落地的Agent系统绕不开四样东西:模型、工具、记忆、控制循环。
模型是推理引擎,负责理解输入、拆解任务、决定下一步该调用什么工具。它不需要多聪明,但必须愿意输出结构化的工具调用指令,否则整个循环推不动。
工具是Agent的"手臂",让它能查数据库、调外部接口、读写文件。没有工具的Agent只是个高级问答盒子,有了工具才真正能干活。
记忆分两层:短期记忆就是当前对话的上下文,长期记忆是跨会话的摘要、用户偏好、历史结果。记忆的质量直接决定Agent会不会"失忆"。
控制循环是骨架,它是一段循环逻辑:把用户问题交给模型,模型说要调用工具,程序执行工具,把结果返回给模型,模型看到结果后继续判断,直到它认为可以给出最终答案为止。
四个部件的关系可以这样理解:模型是大脑,工具是手脚,记忆是记事本,控制循环是那个"每走一步停下来看一眼方向"的机制。后面所有篇幅,都是在把这四个部件一个个做扎实。如果你是零基础,第一遍读可以先不纠结代码,只用这四个词对照着看,思路会清晰很多。之后再去看任何Agent框架、Agent产品,你都能一眼看穿它到底在解决哪个部件的问题。
2. 动手前的脚手架:技术选型逻辑和最小目录结构
2.1 我为什么没有一上来就选重型编排框架
现在的Agent生态里,编排框架和可视化平台非常多,有些确实做得漂亮,拖拽就能搭流程。但我的建议是:第一版一定不要用重型框架,先用最朴素的方式把控制循环手写出来。
原因很简单:Agent的调试难度和抽象层级成正比。你用了框架,等于给模型套了一层壳,一旦出现"模型乱调用工具"或者"上下文丢消息"这类问题,你很难分清是框架的问题、你自己的提示词写得有问题,还是模型的判断出了偏差。手写循环的情况下,每一行消息如何组装、每一步调用如何发生,你全部看得见、改得动。
而且对小白来说,手写一遍循环本身就是最好的学习过程。等你把循环逻辑真正跑通、把踩坑经验沉淀下来,再引入框架提升工程效率完全来得及。从零到生产这条路,第一版不设黑盒比什么都重要。
所以这篇文章里的代码不绑定任何特定框架。你会看到我用chat_model(messages, tools)这样的伪SDK来约定模型调用,用Tool数据结构约定工具形态。你只需要把你实际使用的模型客户端塞进chat_model的位置,其他逻辑原样保留,就能跑起来。
2.2 先把接口约定死,后面才不会被改到怀疑人生
写任何功能之前,先定义两个核心数据结构,这两个是你的地基。
第一个是模型返回结果。我约定一个ModelResponse对象,包含三个字段:
@dataclass class ModelResponse: content: str # 模型输出的文字内容 tool_calls: list[ToolCall] # 模型要求调用的工具列表,可能为空 usage: dict # token消耗信息,生产里一定要看 @dataclass class ToolCall: id: str # 工具调用ID,返还结果时要对齐 name: str # 工具名 arguments: dict # 传给工具的参数,JSON字典第二个是工具注册表。每个工具都是一个Tool对象,包含四样信息:
@dataclass class Tool: name: str # 工具名,建议小写加下划线 description: str # 给模型看的描述,写清楚用途和适用场景 parameters: dict # JSON Schema格式的参数说明 func: Callable # 真正的执行函数这两个接口看起来简单,但它们是后面所有逻辑站稳的前提。tool_call_id必须保留,因为模型可能在同一轮里并发发起多次工具调用,返还结果时必须准确匹配到对应的那一次;description是写给模型看的,不是写给程序员看的,很多人会在这里偷懒,后面必吃苦头。
2.3 一个能一路走到生产的项目目录长这样
目录不需要豪华,但分层的习惯从第一天就要开始养。我常用的最小目录是这样:
agent_app/ ├── main.py # HTTP服务入口 / 命令行入口 ├── config.py # 模型参数、限额、超时、环境配置 ├── agent/ │ ├── core.py # 控制循环:思考-行动-观察 │ ├── prompts.py # 系统提示词集中管理 │ ├── memory.py # 会话记忆:短期缓冲与长期摘要 │ └── tools/ │ ├── registry.py # 工具注册与分发 │ ├── realtime.py # 实时查询类工具 │ └── business.py # 业务类工具 ├── observer.py # 日志、追踪、费用统计 ├── tests/ # 循环逻辑和工具的单元测试 └── requirements.txt我见过太多人把工具函数直接写在main.py里,第一周很爽,第二周就开始痛苦。工具一旦超过五个,注册、路由、测试、回滚都需要清晰的目录边界。这里顺带说一句:observer.py不是可有可无的,后面第五章会专门讲,一个生产级Agent的命脉其实是观测。运行的时候看不到每一轮推理,等出了问题就只能靠猜。
3. 第一版可运行的Agent:最简单的"思考-行动-观察"循环
3.1 定义第一个工具:从"嘴炮"到"能动手"
理论讲再多,不如来一个能跑的。我们做一个最简单的信息查询Agent:用户问天气,Agent能调用天气接口把结果拿回来。先定义工具:
import json def get_weather(city: str, date: str = "今天") -> str: """ 查询指定城市、指定日期的天气情况。 返回JSON包含: city, date, high_temperature, low_temperature, precipitation_probability, weather_desc。 """ # 真实项目里请替换为你的天气API调用,这里用固定返回便于演示 data = { "city": city, "date": date, "high_temperature": 27, "low_temperature": 19, "precipitation_probability": 80, "weather_desc": "中雨转阵雨", } return json.dumps(data, ensure_ascii=False) tool_weather = Tool( name="get_weather", description=( "查询某城市某日天气。适用于用户询问天气、出行准备、穿衣建议等场景。" "date参数支持'今天'、'明天'、'YYYY-MM-DD'。" "查询失败时返回包含success=false和error字段的JSON。" ), parameters={ "type": "object", "properties": { "city": {"type": "string", "description": "城市名,如:青川市"}, "date": {"type": "string", "description": "日期,默认今天"}, }, "required": ["city"], }, func=get_weather, )这里有两个细节建议你死记。第一,description是写给模型看的。你写"获取天气数据"模型也能懂,但如果把适用场景、触发条件、参数约束都写清楚,模型在模糊询问时就知道该不该调用它。第二,工具返回一定是稳定结构,最好直接返回JSON字符串,不要返回Python对象。因为后面要把结果塞回对话上下文喂给模型,字符串永远是最不容易出错的载体。
3.2 核心循环代码:把"思考-行动-观察"变成可运行的结构
接下来是Agent的心脏,控制循环。这里我刻意不用任何框架:
def run_agent(user_input: str, system_prompt: str, tools: list[Tool], max_steps: int = 8): tool_map = {tool.name: tool for tool in tools} messages = [{"role": "system", "content": system_prompt}] messages.append({"role": "user", "content": user_input}) for step in range(max_steps): resp = chat_model( messages=messages, tools=[ {"name": t.name, "description": t.description, "parameters": t.parameters} for t in tools ], tool_choice="auto", ) if not resp.tool_calls: return resp.content messages.append({ "role": "assistant", "content": resp.content, "tool_calls": [ {"id": c.id, "type": "function", "function": {"name": c.name, "arguments": c.arguments}} for c in resp.tool_calls ], }) for call in resp.tool_calls: tool = tool_map.get(call.name) if tool is None: result = json.dumps({"success": False, "error": f"未知工具: {call.name}"}) else: try: result = tool.func(**call.arguments) except Exception as exc: result = json.dumps({"success": False, "error": str(exc)}) messages.append({ "role": "tool", "tool_call_id": call.id, "content": result, }) # 观察结果后,循环继续,让模型决定下一步 log_step(step=step, messages=messages, usage=resp.usage) return "很抱歉,这个问题比较复杂,我连续处理了多步还没能收敛。建议换个说法,或者把需求拆小一些。"这段代码的骨架就是经典的Agent模式:模型说做什么,程序做什么,把结果还回去,模型再看一眼。你可能会问,为什么要有max_steps?因为模型是完全有可能陷入死循环的,比如天气接口一直返回异常,它就可能一直重试。没有步数上限的Agent,轻则烧钱,重则把下游系统打到限流。8步是我比较常用的初始值,业务复杂的可以放到15,但尽量不要更高。
3.3 上下文组装细节:为什么消息顺序这么重要
上面循环里最容易被忽略的是messages这个列表的组装逻辑。它必须严格按时间顺序追加:system提示词在最前,接着是用户输入,然后是模型要求调用工具的那条assistant消息,工具的执行结果以tool角色紧跟其后,下一轮模型根据工具结果做出的新判断再作为新assistant消息出现。
顺序一旦错了,模型就会"失忆"。工具结果放到用户消息之前,模型可能根本不知道这个结果对应哪次查询;漏掉上一轮assistant消息,模型连自己刚才提出要查天气这件事都不记得,只会重复调用同一个工具。这类问题属于"看着简单、踩起来很痛"。
提示:第一版跑通之后,建议专门打印一次完整消息列表,亲眼看看多轮循环之后它长什么样。Debug定位上下文问题时,这条经验能帮你省下至少半天的排查时间。
4. 让Agent真正"实用"的三次改造:记忆、工具编排、边界兜底
第一版能跑,但肯定不好用。你试试连续问三轮问题,第二轮开始它就可能忘了你第一轮说过的关键信息;工具加到十几个,它又开始乱选工具。这章讲实用化的三次改造,都是真实项目里验证过有效的手段。
4.1 记忆怎么落:从进程内字典到"短期缓冲+长期压缩"
本地开发时,很多人把记忆简单实现成一个Python字典:session_id -> messages。我得说这没问题,但你要清楚它的两个致命伤:进程一重启就丢,多实例部署时互相看不见。生产环境的底线是把它放到独立的内存数据存储里,保证无状态Worker都能读到同一个会话。
比存储更棘手的是上下文长度。一个Agent多跑一步,消息数量就会翻着倍涨,用户问题、模型调用、工具结果、模型下一轮判断,四段才算一轮。所以要用两层记忆:
- 短期缓冲:最近N轮消息原样保留,比如6轮以内。
- 长期压缩:超过阈值时,调用模型把旧消息总结成摘要,放在用户消息之前。
简化的压缩逻辑:
def compress_messages(messages, max_turns=6): if len(messages) <= max_turns * 4 + 2: return messages old_part = messages[:-max_turns * 4] text = "\n".join( f"[{m['role']}] {m['content']}" for m in old_part if m.get("content") ) summary = chat_model([ {"role": "system", "content": "请将以下对话历史压缩为300字以内的关键事实摘要,保留用户意图、已查到的结果和未完成事项。"}, {"role": "user", "content": text}, ]).content return [{"role": "system", "content": f"[历史摘要] {summary}"}] + messages[-max_turns * 4:]这个方案不复杂,但非常有效。注意压缩时一定不要漏掉"未完成事项":比如用户上一轮让Agent查了两家酒店的报价,还没决定,压缩时漏了这条,下一轮Agent就会假装这事没发生过,用户体验会非常差。
4.2 工具编排:工具一多,模型就"选择困难"
第一版只有两三个工具时,模型选得很准;等工具加到十五个,模型开始频繁误调用。用户问数据库的问题,它去调了新闻搜索。这几乎是每个Agent项目必经的阶段,不用怀疑是自己能力不行。
解决办法是给工具分域,并在系统提示词里声明路由规则。我习惯分三组:
- 实时查询组:天气、新闻、股票、汇率这类外部信息
- 内部业务组:查订单、改库存、读报表这类系统内部操作
- 系统能力组:计算器、单位换算、日期计算这类通用小工具
然后在提示词里写明:如果用户问题跨多个域,先拆解再行动;如果单一域内工具不明确,先输出判断依据再调用,不要猜测。更彻底的方案是在进入循环之前加一次"意图路由":先用一次轻量模型调用判断问题属于哪个域,再把该域的工具列表传给Agent循环。工具少了,准确率自然会上去。代价是多一次模型调用,但换来的是稳定性,这笔账算得过来。
一个反直觉的经验是:给模型的选项越少,它的选择越可靠。不要把所有工具都塞进上下文,工具不是越多越好。
4.3 防跑飞:超时、幂等、费用限额一个都不能少
生产环境最怕的不是Agent能力不够,而是它不受控制地跑飞,疯狂烧token、反复调用工具、把下游系统打崩。我的防跑飞三板斧:
第一,单次调用和整个循环都要有超时。单次模型调用可以容忍慢,但绝不能无限慢;整个循环更要有外墙。我常用的配置是单次模型调用60秒、整个Agent循环150秒,超时就返回预设的兜底话术。
第二,写操作工具必须幂等。Agent循环一旦重试,写操作可能被执行两遍,用户就多了两个重复订单,这种事故一次就够你记住一整年。最简单的做法是给写操作加一个request_id参数,每次发起时自动生成,下游用它做去重。
第三,按会话设置费用上限。每一步循环都累加token消耗,超过阈值就主动停止,告诉用户"这个问题超出了我可承担的处理成本,请简化后重试"。不要觉得设一块钱太少,失控循环烧掉的钱会远超你的想象,第六章我会算一笔实际的账。
5. 从本地Demo到生产环境:部署链路和可观测性改造
本地跑通和生产可用之间有一条很宽的大河,绝大多数项目是在这里没的。不是说把代码扔到服务器上就叫生产,而是系统能扛住真实用户、真实流量、真实故障。这一章是我做过的最重要的三层改造。
5.1 生产环境跑Agent和本地跑最不一样的三点
本地跑Agent,你是一个人用一个会话,内存里一个dict就够。生产环境面对的是几十上百个并发用户,每个人的会话必须隔离,不能串。这是第一点:会话隔离。本地你只有一个消息列表,生产里每个用户有自己独立的上下文,你不能把所有用户的对话塞进同一个全局变量。
第二点是长任务不阻塞HTTP。一个Agent循环动辄几秒甚至几十秒,如果直接用HTTP同步请求处理,用户那头的请求早超时了,Web服务线程也会一堆堆占满。生产方案是把它放到后台任务队列里,用户提交请求后先拿到一个任务ID,前端轮询或者通过消息通道等结果。
第三点是可观测。本地跑错了可以打印日志看一眼,生产环境出问题你连用户当时的上下文都看不到,没有完整追踪就只能靠猜。这三点是Agent项目上线的"高速公路护栏",不做的话,系统上线满一周就会被迫回头补课。
5.2 无状态化改造:把会话和记忆从进程里搬出去
我建议的第一刀,就是砍掉Agent进程里的所有本地状态。
改造前:Worker进程里挂着一个大字典,存着所有用户的消息列表。 改造后:消息列表存到独立的内存数据存储里,每个会话一个key;任务队列收到新任务再把它读出来,跑完再写回去。这样Worker可以自由扩缩容、滚动重启,任何一个实例挂掉都不会丢会话。
核心流程的伪代码如下:
def handle_task(task): session_id = task["session_id"] messages = kv_store.get(f"session:{session_id}") or [] messages.append({"role": "user", "content": task["user_input"]}) final_reply = run_agent(messages=messages, tools=load_tools(), ...) messages.append({"role": "assistant", "content": final_reply}) kv_store.set(f"session:{session_id}", messages, ttl=60 * 60 * 24) return final_reply注意,Agent循环跑到一半时的中间消息也要及时写回存储。为什么?用户刷新页面、Worker崩溃,都可能让跑了一半的任务彻底丢失,用户还得从头解释需求,体验非常差。你只需要在每轮循环结束时调用一次写入,成本几乎可以忽略,收益却很大。
5.3 记录每一次推理:日志、追踪和费用控制
生产环境的Agent调试,基本靠日志活命。我给每个会话都生成一个trace_id,让用户的请求、后台任务、每一次模型调用、每一次工具执行都带上同一个ID。这样用户报问题时,你只要拿到trace_id,就能把他在那个会话里的所有推理过程拉出来回放。
一条生产日志至少要有这些字段:
| 字段 | 说明 |
|---|---|
| trace_id / session_id | 链路追踪与会话隔离 |
| step | 当前是第几轮循环 |
| role | system / user / assistant / tool |
| message摘要 | 消息过长时截断保存 |
| tool_name与参数 | 调了哪个工具、传了什么参数 |
| tool_result摘要 | 工具返回结果,失败原因必须记录 |
| token用量 | prompt_tokens / completion_tokens |
| 耗时 | 这一步花了多少毫秒 |
| 最终答复/兜底原因 | 是正常收尾还是被强制终止 |
这套日志最大的价值不只是排查bug,更是费用与效果归因。定期跑一次统计:平均每个会话消耗多少token、哪些工具调用占比最高、哪类问题经常走到max_steps兜底。有了这些数据,你才有依据去优化提示词、裁剪工具、调整步数上限,而不是凭感觉乱改。
5.4 上线前发布检查清单
每次Agent项目上线前,我照着这张单子过一遍:
- 会话存储是否独立于进程?重启会不会丢状态?
- Agent任务是否已放入后台队列?HTTP请求会不会被长任务拖死?
- 每个会话的并发是否做了控制?同一个用户连续发两条消息,上下文会不会互相覆盖?
- 模型调用有没有超时、重试、步数上限?
- 工具有没有失败兜底返回?下游异常会不会把脏数据写进会话?
- 是否记录了完整trace日志,能按会话回放全过程?
- 是否按会话或用户设置了费用上限和限流?
- 系统提示词和工具列表是否做了版本管理?改动可回溯?
前面几项偏工程,后面几项偏治理。新项目上线前哪怕只过了前四项,也已经比大多数"能跑"的Demo强了一个层级。
6. 我在真实项目里踩过的坑,和对应的止损办法
最后这部分是我最想分享的。前面讲的是"应该怎么做",这里讲讲我当时"做错了什么"。
6.1 工具返回的脏数据,污染了模型的后续判断
我们早期有个查库存的工具,在查不到时返回{"result": "null"}这样一条看似无害的数据。问题来了:模型看到"null",经常把它当成一个叫"null"的仓库名继续推理,然后一本正经告诉用户"null仓库显示无库存"。这个错误非常隐蔽,提示词、循环、上下文看起来都没问题,纯粹是工具返回值格式设计得不够严谨。
我现在的做法是两条硬规则:工具返回必须带明确的success字段,失败时用error字段说明原因;系统提示词里加一句"工具返回失败时,直接告诉用户暂时无法获取该信息,不要推测"。记住:模型相信工具返回的任何字符串,脏数据就是脏逻辑。
6.2 多轮会话里上下文越攒越脏,Agent开始自己编历史
项目早期没有做记忆压缩,把所有消息一直堆在上下文里。跑到第十几轮,模型开始出幻觉:把前面某次工具调用失败记成了成功,把用户已经放弃的方案当成正在执行。原因也很好理解,旧对话里的失败信息、模棱两可的表述一直在上下文里反复搅动。
止损办法就是4.1里的压缩和剪枝。但我要额外提一个点:压缩摘要时一定要带上"已放弃方案",否则模型容易在后续轮次重新提出用户已经否掉的方案,显得特别没记性。这个细节我在第一次压摘要时漏掉了,结果线上用户跟Agent在一个问题上兜了好几圈,因为它第三次推荐了同一个已经被否定的选项。
6.3 用户一句话触发了十分钟的Agent马拉松
有一次用户问了一个跨多个工具的大问题,我们没有设费用上限,Agent在循环里连续调用了将近二十次工具,前后跑了快十分钟,token烧了一大把。用户早就在页面上关掉了,留下一张离谱的费用账单。这次的教训很直接:Agent不是越能跑越好,能及时停下是更稀缺的能力。
现在的兜底策略是:三步以内搞不定的复杂需求,Agent会主动向用户确认优先级和范围,而不是默默一路狂奔;同时max_steps和总耗时限制写死在配置中心,任何改动都要走审批。
6.4 成本失控要算总账,不能只看单次
很多人只关注"一次调用多少钱",忽略了一个Agent循环里会有多少步。我统计过我们早期某个Agent,平均一个会话要调模型8到10次,加上工具调用,token消耗是单纯问答场景的十倍以上。单次价格再低,乘以十再乘以日活用户,账单就会非常难看。
我的止损方案是在观测层把费用折算成每会话成本,每天出报表,超过阈值当天告警。同时把一些高频、结果稳定的工具做缓存,比如天气查询一小时内的结果直接复用,避免模型重复调用。省下来的钱是一方面,更重要的是让团队重新把注意力放回"这个Agent到底值不值"这个问题上。
说实话,这些坑没有一个是"技术太难"造成的,全是工程意识不到位。Agent的代码本身不难,难的是你愿意在它失控之前设好边界,在出问题时能快速看到过程。这也是我一直强调手写循环、强调可观测的原因——你看到的每一步,才是你最终能控制的每一步。
按这条路径走下来,你大概率会经历几个阶段:头三天觉得"不过如此",一周后开始撞上上下文污染,半个月后突然理解为什么要有上限和兜底。我个人的体会是,Agent系统真正的门槛不在"让它会",而在"让它收敛、可追溯、敢上线"。所以脚手架搭好之后,先别急着加工具,去接一个真实的数据源跑一周,把日志和坑记录下来。这条闭环跑完,你再看各种框架和平台,就再也不会被术语绕晕了。祝你的第一个Agent早点收口上线。