企业官网建设流程全解析

这个系列写到第三篇，我想换个节奏：不再手把手教你怎么跑通一个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方案不算少，我按实际经验做个横向对比，你就不用挨个趟雷了。

如果你只是做原型验证，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_factory=list) 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_ascii=False) 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_defs=list(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_model=None): 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", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\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 高频问题速查表

我把实际项目里最常遇到的问题整理成了一张表，按“现象、可能原因、处置办法”三列排列。你可以把这张表存下来，上线前照着检查一遍。

6.2 几条踩出来的避坑心得

先说最大的一条：工具调用是成本最大来源，也是最容易失控的环节。很多新手只盯着模型输出质量，忽略了工具返回体的大小和调用次数。我建议在Agent循环里显式统计每个工具调用消耗的Token，做到心中有数；超出预算阈值就中断当前任务或自动降级。

第二条是不要过早抽象。我第一次做Agent时，花了很大精力设计可插拔的架构，结果大部分抽象派不上用场，反而让调试变得困难。现在我的做法是先用最简单的循环快速验证逻辑，确认任务闭环跑通后，再用状态机、队列、缓存逐层加固。顺序很重要，反过来的成本高得多。

第三条是每次修改模型版本或提示词核心部分，都要跑一遍回归样例集。这个集合不用大，20到50条覆盖主要业务场景就够。跑完对比输出差异，重点看有没有违背规则的内容。没有回归集，你怎么知道这次改提示词没有把上一个功能修坏。

最后分享一个小技巧：给Agent定义一个“结果无变化即停止”的规则。如果某轮工具返回结果和上一轮完全一样，说明模型已经在原地打转，此时强行继续只会烧钱。我加了这个判断后，线上Agent的无效循环次数明显下降，成本也稳住了。这个思路在各家框架里没有统一实现，自己动手写下很快，收益却立竿见影。

这个系列已经写到第三篇，从架构选型、Token管理、状态机、工具循环、并发调度的演示到部署运维，基本把Agent从Demo到生产的完整路径走了一遍。我自己在真实项目里最大的体会是：Agent能不能用，取决于你在模型之外建了多少约束。模型负责聪明，工程负责稳定。后面如果你们想继续深挖，可以再聊RAG增强、向量库选型，或者某个具体行业场景的Agent落地案例，这些内容我都有实战记录，等你们的反馈吧。