这个系列写到第三篇我想换个节奏不再手把手教你怎么跑通一个demo而是聊点真正把Agent推向可用状态时才会遇到的东西。前两篇主要覆盖了Python环境、模型调用、基础工具函数这些都是地基但从“能跑”到“能稳定跑”中间隔着状态管理、Token成本、并发调度、部署运维这几座山。这篇文章就围绕Python AI Agent开发中这些进阶问题展开所有内容都来自我自己在真实项目中踩过的坑和验证过的方案。适合刚跑通第一个Agent、准备把它接到业务里或上线的同学参考也适合想搞清楚Agent底层工作机制的读者。放心源码级细节和可直接抄的代码都会有而且我会把为什么这样做讲透。1. 认清Agent的主流架构再决定怎么写代码1.1 三类主流架构从单循环到多智能体现在业界聊Agent架构各家白皮书和开源项目虽然叫法不同但底层基本收敛到三类。第一类ReAct风格的单循环。模型在“思考-行动-观察-再思考”之间循环直到拿到完整答案。AutoGPT、早期的BabyAGI很多还记得名字的项目核心就是这个。这类架构最大的优点是透明可控每一步干了什么、为什么这么干日志里看得清清楚楚缺点是随着上下文增长模型容易忘掉前面的约束错误会被后面的步骤放大。适用场景是工具调用多、任务能按顺序拆着走的比如网页抓取、信息查询、文档生成工作流。第二类规划器加执行器。规划Agent负责理解任务、拆解步骤、下发指令执行器只负责具体干活可以是子Agent也可以是一组工具函数。这个模式适合业务流程固定、边界清晰的场景比如客服工单流转规划Agent判断工单类型然后调用对应系统接口最后统一汇总回复。好处是隔离了“思考”和“执行”一个地方出错不会连坐整个流程。第三类多智能体协作。核心思路是模拟团队成员各有角色彼此通过消息交换结果。国内外的MetaGPT、CrewAI、AutoGen都在往这个方向使劲。适合真正需要多种专业能力协作的任务比如产品需求到代码落地、复杂研究报告生成。缺点也很明显Token消耗指数级上升错误会在Agent之间传播排查起来极其痛苦。怎么选我的建议很直接先单循环跑到业务闭环再根据瓶颈判断要不要拆成多Agent。很多项目最初根本不需要多Agent硬上只会多花钱多踩坑后面我会专门讲这个问题。1.2 Python框架选型与取舍从“自研”到“全家桶”确定架构之后第二个问题是选框架还是自己写。目前主流的Python方案不算少我按实际经验做个横向对比你就不用挨个趟雷了。框架核心抽象适合场景上手难度常见坑LangGraph图状态机流程可控、需要条件分支/人机协同中抽象层多报错信息绕AutoGen多Agent对话两个或多个角色协作、自动对话低对话轮数容易失控CrewAI角色任务模拟团队分工、任务编排低复杂流程表达能力有限LlamaIndex Workflow事件驱动工作流RAG为主、事件流处理中和Agent职责边界模糊自研状态机工具注册表自己定义生产环境、深度定制高需要自己处理很多东西如果你只是做原型验证CrewAI这类框架最快如果你的业务分支多、需要精确控制流程LangGraph更合适但我个人在生产项目里反而倾向于自研一个最小的回调循环大概200到300行就能覆盖需求。原因不复杂框架带来的抽象一层套一层等你要加缓存、限流、审计日志的时候框架又成了障碍。顺带说一句搜索热词里“基于Rust语言AI Agent”也不少见。Rust确实在性能和并发安全上有优势编译期强制处理很多并发问题适合做工具执行引擎、API网关这类对资源敏感的基础设施。但AI Agent业务逻辑迭代太快Python在模型调用、数据清洗、生态工具链上的优势太明显了所以我的选择是业务层用Python性能瓶颈再用Rust补位而不是一上来就用Rust写全部业务。2. 把Token机制当成水电煤不懂它很难控成本2.1 Token到底怎么算为什么Agent这么费Token关于“AI Agent token是什么意思”这是很多新手问得最多的问题。Token是模型处理文本的基本单位既不是字符也不是单词而是经过分词器切分后的最小文本片段。英文场景下一个常见单词大约占1到1.5个Token中文场景下一个字通常占1到2个Token。模型计费和上下文窗口都以Token为计量单位。Agent之所以特别费Token是因为一次任务并不是一次模型调用而是一个多次调用的循环。我给了个具体数字你感受一下。假设你的系统提示写了800 Token工作记忆1200 Token某次工具返回1500 Token用户任务200 Token单次输入就接近3700 Token模型写回答或工具调用参数输出平均按800 Token算。如果一次任务需要25次“思考-行动-观察”循环那这一个任务的调用量就是25次满天飞。一天100个任务累计输入超过9百万Token输出超过2百万Token。单Agent任务的数量乘以循环次数成本就是这么被放大的。价格参考方面各家模型差异非常大目前主流大模型大致在输入2到3美元/百万Token、输出10到15美元/百万Token这个区间也有便宜一个数量级的国产模型和开源模型。具体价格会变我不写死但你套上面那个量级就能算出一天不加优化跑掉几十美元是很容易的事。所以Token优化不是可选动作是Agent开发的基础能力。2.2 压缩Token的四种实用手段第一招给System Prompt做减法。很多新手喜欢把规则、案例、约束全部堆进系统提示觉得提示越长越严谨。实际效果恰恰相反提示越长模型的有效注意力越分散推理速度也越慢。我建议系统提示只保留不可动摇的规则和角色定位可选项尽量外置到用户消息里按需注入。这一点在热词里没有直接体现但这是我做过的优化中收益最明显的一项。第二招历史消息做滑动窗口加摘要。不要把所有历史会话都塞给模型模型没有“无限记忆”也没有必要记那么清楚。通常保留最近N轮完整消息更早的内容由模型生成一段摘要每次请求时拼上这段摘要即可。N的大小可以根据任务复杂度调初步建议10轮起步。第三招工具返回只取必要字段。工具调用返回结果不要整个JSON原文丢给模型而是加工成模型真正需要的那几个字段。比如天气接口返回几十个字段模型只需要天气和温度那就只保留这两项。大字段写不进上下文还容易让模型在无关数据里“走神”。第四招用小模型做路由和摘要。一次Agent循环里不是每一环都需要最强模型。意图判断、文本分类、摘要这类简单任务用便宜的小模型就能完成等到了关键决策点再上最强模型。这样整体成本往往能省50%以上质量几乎不掉。我在生产环境里就是这么做的实测下来很稳。3. Python Agent核心实战一状态机、工具调用与长上下文3.1 用状态机给Agent的一生加约束第一眼看Agent很多人会写一个while循环模型说没完成就继续下一轮。Demo阶段没问题生产环境你不加状态约束很快会被“无限循环”教做人模型会反复调用同一个工具或者在一个死结里来回打转花费全是你的钱。所以我在项目里都会给Agent套一个简单的状态机。核心状态先定义清楚IDLE空闲、PLANNING规划、EXECUTING执行、OBSERVING观察、FINISHED完成、FAILED失败。每一步只允许按预设转移表跳转非法转移直接报错。我写了一个最小的运行时可以直接参考。from enum import Enum from dataclasses import dataclass, field from typing import List class AgentState(str, Enum): IDLE idle PLANNING planning EXECUTING executing OBSERVING observing FINISHED finished FAILED failed dataclass class AgentRuntime: state: AgentState AgentState.IDLE iter_count: int 0 max_iterations: int 20 trace: List[dict] field(default_factorylist) def transition(self, next_state: AgentState) - bool: allowed { AgentState.IDLE: {AgentState.PLANNING}, AgentState.PLANNING: {AgentState.EXECUTING, AgentState.FINISHED}, AgentState.EXECUTING: {AgentState.OBSERVING, AgentState.FAILED}, AgentState.OBSERVING: {AgentState.PLANNING, AgentState.FINISHED, AgentState.FAILED}, } if next_state in allowed[self.state]: self.state next_state return True return False这段代码的重点不是怎么定义枚举而是帮你想清楚Agent每一步的输入、输出、失败条件、终止条件都必须提前约定。max_iterations这个参数很重要我建议生产环境设置在20以内任务特别复杂的再放宽。还有一个容易被忽略的点状态机要有持久化能力。进程一重启当前批次的任务状态要能从数据库恢复而不是全部从头再跑。3.2 工具调用的底层机制先手写一遍循环现在很多教程都是基于框架的agent_execute()一键完成但如果你不清楚工具调用底层发生了什么遇到问题完全没法排查。我建议至少手写一遍工具调用循环把链路走通后面再用框架心里才踏实。大致的机制是这样的模型在收到用户消息和工具列表之后如果判断需要调用某个函数会返回一段结构化JSON里面包含函数名、参数以及参数值。你的代码收到之后不是直接把JSON交给模型而是解析它、在本地工具注册表里找到对应函数、传参执行再把工具返回结果作为一条新的消息返回给模型。模型继续判断下一步是再调工具还是给出最终答案。import json from typing import Callable, Dict, Any TOOL_REGISTRY: Dict[str, Callable[..., Any]] {} def register_tool(name: str): def decorator(func): TOOL_REGISTRY[name] func return func return decorator register_tool(get_weather) def get_weather(city: str, date: str today) - str: # 真实项目里这里去请求天气API return json.dumps({city: city, date: date, summary: 晴, 22℃}, ensure_asciiFalse) def run_agent_loop(task: str, max_rounds: int 10): messages [{role: user, content: task}] for _ in range(max_rounds): response call_llm(messages, tool_defslist(TOOL_REGISTRY.keys())) messages.append(response) if response.tool_calls: for tool_call in response.tool_calls: fn TOOL_REGISTRY[tool_call.name] result fn(**json.loads(tool_call.arguments)) messages.append({role: tool, tool_call_id: tool_call.id, content: result}) elif response.stop_reason stop: return response.content raise RuntimeError(max rounds exceeded)这个伪代码里我刻意忽略了call_llm的具体实现因为它跟各家SDK绑定。你要掌握的是这个结构工具注册、参数解析、结果回传、循环终止四件事缺一不可。生产环境我会把工具描述定为JSON Schema格式用Pydantic定义参数类型再用模型的tools参数传给模型。手写一遍这个循环你对“Agent为什么会乱调用工具”的理解会深得多。3.3 长上下文和记忆该忘的就得忘长上下文管理是Agent状态管理里的另一个大坑。模型上下文窗口再大也有上限而且上下文越长单次调用越慢越贵。很多Agent跑着跑着就报context length exceeded几乎都是因为把历史消息一股脑塞进去。我的做法分两层。短期记忆用滑动窗口加摘要保留最近N轮完整交互更早的内容压缩成摘要长期记忆则把多轮对话中有价值的知识点提取出来存进向量库下次任务开始时做相似度检索再注入上下文这就是RAG。两个逻辑看起来简单却能把上下文控制在合理范围之内。class SlidingWindowMemory: def __init__(self, keep_rounds: int 10, summarize_modelNone): self.keep_rounds keep_rounds self.recent_messages [] self.summary self.summarize_model summarize_model def add(self, message: dict): self.recent_messages.append(message) while len(self.recent_messages) self.keep_rounds * 2: expired self.recent_messages[:2] self.recent_messages self.recent_messages[2:] if self.summarize_model: self.summary self.summarize_model( f把这段对话浓缩成两句话{expired} ) def build_prompt(self) - str: prefix f历史摘要{self.summary}\n if self.summary else return prefix \n.join( f{m[role]}: {m[content]} for m in self.recent_messages )这里要注意一个细节摘要本身也需要Token如果对话特别长摘要可能会反过来占据很大比例。所以我会控制摘要的产出长度比如固定输出三到五句话并且用便宜的小模型生成不让摘要的成本把省下的钱又吃回去。4. Python Agent核心实战二并发调度与多Agent协作4.1 什么场景需要多Agent什么场景真不需要搜索热词里“ai agent搭建”持续有人搜说明很多人正处在准备搭建但方向未定的阶段。我经常收到的问题是要不要上多Agent答案是取决于你的任务结构而不是潮流。我列个判断标准如果你的任务可以被拆成多个角色视角、多个独立步骤且每个步骤之间没有强依赖多Agent是合适的比如“产品经理需求-架构师设计-程序员编码”这种模拟团队流程如果任务本身是个线性流程几个工具调用就能走完硬拆多Agent纯属增加成本和复杂度。另一个信号是单Agent的上下文已经塞不下所需信息拆成多个有专业分工的Agent反而能各管一段。多Agent不是没有代价每次Agent之间交换消息都要消耗Token错误会在传递过程中被放大调试链路也会变长。我的建议是先把业务用单Agent跑通跑出一个稳定基线之后再按痛点拆。跳过这一步直接上多Agent很容易发现能力没提升账单先翻倍。4.2 队列、并发与隔离别让一个任务拖垮全站多Agent系统本质上是一个异步任务系统。我见过不少人在Web框架里直接起线程跑Agent任务一多进程就被占满其他请求全部排队。正确做法是解耦任务先写进队列Worker进程按照自己的节奏消费队列消息里带上任务ID和上下文处理完再异步更新状态。import asyncio from asyncio import Semaphore, Queue async def worker(worker_id: int, task_queue: Queue, sem: Semaphore): while True: task await task_queue.get() async with sem: try: result await run_agent_task(task) except Exception as exc: result {status: failed, error: str(exc)} await save_result(task[id], result) task_queue.task_done()代码里这个Semaphore是关键的限流手段。各家模型API都有并发限制不控流会出现大量429报错。经验值是先用模型服务的RPM上限估算如果限速每分钟1000次请求一个任务平均需要10次调用那么同时运行的任务数不要超过100留出余量再打折。限流不是NetOps的事是Agent开发的一部分。再有就是状态隔离。多个Worker并发跑Agent时绝对不要使用全局变量保存某个任务的状态。每个任务的状态应该独立存储要么挂在任务对象上要么落地到数据库。看起来是常识但很多自研实现就死在这上面。数据库的起步方案很轻SQLite就够等任务量上来再换PostgreSQL。4.3 可观测性Agent没有可视化日志等于裸奔Agent是一个多步、动态决策的系统出问题时你很难靠“感觉”判断是哪一步出了问题。所以从第一天起就要把每次LLM调用的信息记录下来任务ID、调用序号、模型名称、输入输出Token数、工具名、工具参数、耗时、是否重试。我习惯把每一轮循环追加到JSON Lines文件里一行一次调用排查问题用grep就能定位到具体某个环节。import json import time import uuid def log_llm_call(context: dict, response: dict, tool_run: dict None): record { ts: time.time(), trace_id: context.get(trace_id, uuid.uuid4().hex), task_id: context.get(task_id), model: context.get(model), prompt_tokens: response.get(usage, {}).get(prompt_tokens), completion_tokens: response.get(usage, {}).get(completion_tokens), latency_ms: context.get(latency_ms), tool_name: tool_run.get(name) if tool_run else None, tool_args: tool_run.get(args) if tool_run else None, error: context.get(error), } with open(agent_trace.jsonl, a, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \n)如果项目规模再大可以把这个trace对接OpenTelemetry或LangSmith这样的工具但我建议先保证自己的日志格式干净统一再谈集成平台。有了日志你才能回答关于“Agent为什么卡住、为什么费用飙升、为什么结果飘忽”这类问题。没有日志的Agent就像记账不做流水出了事只能拍脑袋。5. 部署与稳定运行从开发机到生产的一公里5.1 锁住环境、密钥与模型版本先聊开发环境这个热词话题。你如果本地用虚拟机或者本机多端口跑多个站点要给Agent服务的LLM回调地址配上正确的域名和端口转发否则回调地址不一致会让鉴权失效。我见过不少人在本地调得好好的一上服务器就报签名失败排查到最后发现是回调URL没有同步修改。生产部署第一件事是锁定环境用requirements.txt或pyproject.toml锁定Python依赖版本优先用Python 3.10以上的稳定版本模型版本也要锁不要默认“最新”因为模型厂商升级后行为可能变化今天测得好好的流程明天可能因为模型更新而产出不同结果这会让回归测试形同虚设。密钥管理的教训值得单独说。API Key放在代码里然后误提交到公开仓库这种事至今还在反复出现。本地用.env文件加载并在.gitignore里忽略它生产环境通过环境变量或密钥服务注入。不要把密钥打进镜像也不要在日志里打印消息体特别是涉及业务数据的时候日志脱敏规则要在开发阶段就定下来。5.2 重试、缓存与限流让Agent抗住真实流量线上和开发机的最大区别在于不确定性模型API可能短暂抖动、网络可能闪断、超时时有发生。所以重试策略必须有但要克制。不要无脑无限重试那会给模型服务造成更大压力。我通常用指数退避最多重试两次三次还失败就直接标记任务失败并告警让人工介入。缓存的价值经常被低估。对工具的确定性返回结果做哈希缓存能显著降低Token消耗同一个城市的天气查询一小时内完全没必要重复调用两次模型。但要判断缓存是否适用于当前场景结果有时效性、涉及私有数据时不要盲目缓存。系统提示这类稳定文本尽量复用模型服务商提供的前缀缓存能力也能省下一大截。限流和重试是一对搭档限流做在前面重试兜底在后。用户不关心底层模型怎么调但你要保证流量上来时任务不会打爆限额。5.3 状态持久化与数据落地Agent任务一旦进入生产就不能拿内存里的状态当账本。任务状态和中间结果需要落到数据库里至少要有张任务表和一张运行轨迹表。PostgreSQL的JSONB列、MySQL的JSON类型都行起步阶段SQLite也够用。字段建议包含任务ID、状态、当前步骤、输入摘要、输出结果、Token总消耗、创建时间、更新时间。写出任务表很简单但生产环境大多数人会忽略“Token总消耗”字段。不记录这个字段月底账单来的时候你根本不知道钱花在哪。另外不管是日志还是数据库用户输入和模型输出都属于业务数据该脱敏的字段必须脱敏。Agent开发做得越深越要明白一个道理输出质量重要安全性更重要。不是鼓励你层层加审批而是至少保证密钥不被打印、敏感信息不落到明文日志里。6. 高频问题速查与避坑心得6.1 高频问题速查表我把实际项目里最常遇到的问题整理成了一张表按“现象、可能原因、处置办法”三列排列。你可以把这张表存下来上线前照着检查一遍。现象可能原因处置办法报context length exceeded历史消息全量塞入上下文滑动窗口加摘要控制单次输入Token模型反复调用同一个工具缺少终止条件或工具结果未变化加最大轮数工具返回前先做去重判断工具调用参数格式错误参数Schema不清晰用Pydantic定义参数生成JSON Schema传入模型单个任务响应很慢历史过长、模型过重精简消息意图路由阶段换小模型费用一个月下来吓人无缓存、工具返回过大、循环次数多缓存去重工具返回精简设置轮数上限并发一高就报429超过模型API限速信号量限流指数退避重试降低同时运行任务数进程重启任务全部丢失状态没持久化任务状态和轨迹落库启动时恢复未完成任务同一输入结果不稳定temperature偏高或模型版本变动调低temperature锁定模型版本6.2 几条踩出来的避坑心得先说最大的一条工具调用是成本最大来源也是最容易失控的环节。很多新手只盯着模型输出质量忽略了工具返回体的大小和调用次数。我建议在Agent循环里显式统计每个工具调用消耗的Token做到心中有数超出预算阈值就中断当前任务或自动降级。第二条是不要过早抽象。我第一次做Agent时花了很大精力设计可插拔的架构结果大部分抽象派不上用场反而让调试变得困难。现在我的做法是先用最简单的循环快速验证逻辑确认任务闭环跑通后再用状态机、队列、缓存逐层加固。顺序很重要反过来的成本高得多。第三条是每次修改模型版本或提示词核心部分都要跑一遍回归样例集。这个集合不用大20到50条覆盖主要业务场景就够。跑完对比输出差异重点看有没有违背规则的内容。没有回归集你怎么知道这次改提示词没有把上一个功能修坏。最后分享一个小技巧给Agent定义一个“结果无变化即停止”的规则。如果某轮工具返回结果和上一轮完全一样说明模型已经在原地打转此时强行继续只会烧钱。我加了这个判断后线上Agent的无效循环次数明显下降成本也稳住了。这个思路在各家框架里没有统一实现自己动手写下很快收益却立竿见影。这个系列已经写到第三篇从架构选型、Token管理、状态机、工具循环、并发调度的演示到部署运维基本把Agent从Demo到生产的完整路径走了一遍。我自己在真实项目里最大的体会是Agent能不能用取决于你在模型之外建了多少约束。模型负责聪明工程负责稳定。后面如果你们想继续深挖可以再聊RAG增强、向量库选型或者某个具体行业场景的Agent落地案例这些内容我都有实战记录等你们的反馈吧。