这几年只要聊到AI应用,Agent几乎绕不开。我见过不少团队第一步就栽在架构选型上——PPT里的Agent无所不能,代码里却不知道该把“规划”放哪里、“记忆”放哪里、工具调用的结果谁来解析、任务什么时候算结束。其实很多人没意识到,网上流传的“Agent七要素”根本不是什么学术黑话,而是一份现成的工程需求清单。把这七个要素翻译成代码模块,再把运行时的关键选择点拎出来,Agent的工程实现就没有那么玄。
这篇文章适合两类人:一类是准备把Agent接进业务系统的后端工程师,另一类是已经在用LangGraph、Spring AI甚至Rust折腾Agent,但总觉得差点意思的实践者。我会从七要素讲到七个决策点,再聊框架选型、并发瓶颈和Token成本,最后给一份FastAPI + LangGraph的服务化实现骨架,并结合最近讨论度很高的几个场景说说下水感受。
1. 别把七要素当概念,它是你写给自己的工程需求清单
1.1 Agent七要素到底指什么
业内聊Agent,最常见的提法是目标、大模型、规划、记忆、工具、执行和反馈、终止与评估。不同文章叫法略有差别,但核心逃不出这七样。听起来很理论,实际上每一要素都对应一段必须有人写的代码:
- 任务与目标输入:Agent要解决什么问题、输入是什么格式、输出给谁看。
- 大模型推理决策:核心的“大脑”,负责理解上下文、做判断、生成内容和选择下一步动作。
- 规划拆解:把一个复杂目标拆成若干可执行步骤,决定先做什么、后做什么。
- 记忆管理:短期记忆保存当前会话上下文,长期记忆保存用户偏好、历史事实或领域知识。
- 工具调用:Agent能访问的外部能力,比如查数据库、调订单接口、发消息、算价格。
- 执行与反馈闭环:调用工具后拿到结果,把结果解析、结构化、判断是否合理,再喂回给模型。
- 终止与结果评估:判断任务是否完成,输出是否满足要求,以及失败时如何收场。
我见过太多项目,把这七要素画在了架构图里,但代码里真正存在的只有一个入口、一个模型调用和一个巨大的System Prompt。短期demo能跑,一旦工具数量超过三个,Prompt里编排逻辑就会变得又臭又长,模型输出稍微不稳定,整个链路就崩。
1.2 每个要素在工程上对应什么
七要素翻译成工程模块,大概是下面这张表:
| 七要素 | 工程模块 | 代码落点 |
|---|---|---|
| 任务与目标输入 | 请求入口与意图解析 | API入参、System Prompt、结构化输入校验 |
| 大模型推理 | LLM调用层 | ChatOpenAI、Spring AI ChatClient、模型网关 |
| 规划拆解 | 规划器/任务拆解 | 状态图节点、ReAct循环、Plan-and-Execute |
| 记忆管理 | 上下文存储 | 消息列表、Redis临时态、向量数据库 |
| 工具调用 | 工具注册与执行器 | Tool Schema、Function Calling、工具函数 |
| 执行与反馈 | 结果解析与状态更新 | 条件边、状态合并、错误捕获与重试 |
| 终止与评估 | 退出条件与校验 | 结束节点、人工审批、输出校验器 |
这里最容易被忽略的是“规划拆解”和“终止与评估”。前者决定了Agent是一口气生成回答,还是真的在“做事”;后者决定了它会不会在一个死循环里烧token烧到爆。
我自己的习惯是:先把这七要素写成一段话,比如“用户下单后,Agent需要确认订单信息、查库存、调用支付接口、返回结果;如果支付失败,要解释原因并给出替代方案”,然后再决定哪些用模型、哪些用规则、哪些用代码硬逻辑。这比先选框架再填业务要顺得多。
2. 七个决策点才是主循环的本体:每轮循环都在做选择题
2.1 主循环不是 while True
很多人理解Agent主循环,就是“让模型反复思考直到输出结束”,于是代码里真的写了:
while not finished: response = model.invoke(messages) # 解析是否需要工具...这种写法不是不行,而是把“决策”全塞给了模型。模型说“我需要工具”,你才调工具;模型说“我完成了”,你才结束。问题在于:模型偶尔会自以为是,工具调用失败后它也经常不会主动修正,于是你被迫在循环里写一堆奇怪的分支判断,最后代码比业务逻辑还难维护。
更工程化的做法,是把每轮循环看成有限状态机里的一次状态迁移。Agent在每个节点只做一件具体的事:判断意图、选择工具、执行工具、校验结果、决定下一步。这正好对应了我常说的“七个决策点”。
2.2 七个决策点逐个拆解
以我最近实现的一个客服Agent为例,每轮执行中至少会出现下面七个选择题:
- 意图是否清晰:用户说“我想退货”,但没给订单号,这时候是直接查还是先追问?
- 是否需要记忆或外部知识:这个问题能不能靠当前对话回答?要不要检索用户历史订单、商品知识库?
- 该调用哪个工具、传什么参数:决定工具名称和入参,这一步通常是模型负责的。
- 工具返回值是否合法:订单API返回了404,是订单号错了还是服务异常?要不要换个工具或重试?
- 信息是否足够结束:拿到工具结果后,能不能生成最终答复?还是需要再追问、再查一个工具?
- 输出是否需要人工确认:涉及退款、发消息、交易操作时,是不是要先走人工审批?
- 失败与超时如何兜底:模型超时、工具超时、连续失败,有没有降级话术或转人工通道?
这七个决策点不一定每轮都出现,但它们必须被显式设计在代码里。比如LangGraph里,每个决策点就是一条条件边:根据状态字段决定下一个节点是“追问”、“调用工具”还是“结束”。
2.3 决策点之间靠状态流转,不靠模型自觉
把七要素和七个决策点放一起看,关系其实很清晰:
| 决策点 | 对应要素 | 判断方式 |
|---|---|---|
| 意图是否清晰 | 任务/目标 | 规则引擎或轻量模型分类 |
| 是否需要记忆/知识 | 记忆管理 | 向量检索命中率、上下文长度判断 |
| 调用哪个工具 | 工具调用/规划 | 结构化输出、Function Calling |
| 工具结果是否合法 | 执行与反馈 | 异常捕获、Schema校验 |
| 是否继续循环 | 终止与评估 | 最大轮数、信息完备度判断 |
| 是否需要人工确认 | 终止与评估 | 业务规则(金额、动作敏感度) |
| 失败如何兜底 | 执行与反馈 | 超时设置、重试策略、降级话术 |
这套设计的好处是,Agent的行为可以被测试、被观测、被限制。你可以给每个决策点加日志,打印“为什么走了这个分支”,出问题直接定位到具体节点。而如果全指望模型在一条Prompt里完成所有决策,那相当于把线上稳定性押注在一段自然语言上,风险太高。
3. 框架、并发与Token:工程选型里的三个真实瓶颈
3.1 框架选型:LangGraph、Spring AI与Rust生态的真实差异
先说结论:没有最好,只有团队最熟。
- LangChain / LangGraph:Python团队上手快,状态图编排直观,适合快速验证和复杂流程编排。缺点是你得接受它迭代快、API变动频繁带来的“追版本”成本。
- Spring AI:Java/Spring团队集成成本低,企业级基础设施很全(配置、监控、AOP),但生态和社区资料目前比LangChain少一些。
- Rust + 自研状态机:并发模型确实强,适合做高吞吐的Agent网关、统一模型接入层这类基础设施。但如果你只是想快速把业务跑起来,用Rust从零搭Agent会痛到你怀疑人生。
- 低代码平台(类似扣子Coze这类):拖拽节点做demo极快,适合运营自建话术机器人,但遇到复杂权限、私有化部署、深度定制时天花板明显。
| 维度 | LangGraph | Spring AI | Rust自研 | 低代码平台 |
|---|---|---|---|---|
| 状态编排 | 强 | 中 | 自研成本高 | 可视化,但灵活度低 |
| 团队门槛 | Python低 | Java中等 | Rust高 | 最低 |
| 并发吞吐 | 依赖部署 | 依赖线程模型 | 高 | 平台决定 |
| 自定义能力 | 高 | 高 | 最高 | 低 |
| 适合场景 | 快速落地、复杂流程 | 企业Java栈 | 模型网关、基础设施 | 原型验证、运营自助 |
最近经常看到“基于Rust语言的AI Agent”这类讨论,说实话,Rust更适合的场景是“Agent的底座”:协议解析、多路并发转发、token计数与流控、模型调用网关。真正业务编排放在Rust里写,除非团队全是Rust老手,否则迭代速度会拖垮你。
3.2 Agent到底该怎么扛并发
这个问题我自己踩过不少坑。Agent的“并发”和普通Web接口的并发是两回事。普通接口是短请求,几百毫秒返回就完事;Agent任务是长链路,一轮任务里要调好几次LLM、好几次工具。假设每个任务要调3次模型(每次约2秒)和2次工具(每次约0.5秒),单任务耗时差不多7秒。
想支撑10 QPS的稳定吞吐,意味着同一时刻至少有70个任务在途。这个数字不是靠“把FastAPI改成异步”就能解决的,而是要把任务从HTTP线程里搬到后台队列里跑,并且对外部调用做严格的控制。
我实际用的方案是:短任务同步返回,长任务异步化。所有Agent任务都带着session_id和task_id,接收后丢进Redis队列,由worker执行,执行结果写回,前端轮询或WebSocket通知。这样HTTP层只负责接收和查询,真正的Agent逻辑在worker里跑,比较方便控制并发上限和重试策略。
几个经验值:
- 并发上限不要拍脑袋定,用
压测QPS × 单任务平均耗时算出在途任务数,再乘1.5到2的冗余系数。 - 所有外部调用必须设超时,LLM和工具都设,宁可返回降级话术,也不要让任务卡死在等待上。
- 工具调用做成幂等的,重试的时候才不会重复下单、重复发送消息。
- 会话状态要能断点恢复,worker重启后任务能从最近节点继续跑,而不是从头再来。
3.3 Token成本为什么是Agent的隐形天花板
很多团队把Agent写通之后,第二天就被账单吓到了。Agent本身就是Token消耗大户,原因很简单:它每轮循环都要把系统Prompt、历史消息、工具返回结果重新发给模型,多轮下来总量轻松破万。
举个例子:系统Prompt 800 token,历史消息平均2000 token,工具返回1500 token,Agent平均跑5轮,一次任务光输入Token就是(800 + 2000 + 1500) × 5 = 21500 token,这还没算生成输出。如果一天跑几千个任务,成本就非常可观。
我从成本角度给的优化建议是:
- 模型分级:规划、推理、总结用强模型;意图分类、实体提取、格式化输出这类重复任务用轻量模型,甚至可以本地搞定。
- 不要无脑把工具结果塞进上下文:工具返回一大段JSON,先抽取出关键字段再喂给模型,剩下的一律丢弃。
- 消息压缩:超过一定轮数后,把旧消息摘要成一段话,而不是无限追加。
- 命中缓存:相同或相似的上下文请求直接用缓存结果,尤其是FAQ类任务,能省下一大笔重复调用的钱。
Token不只是成本问题,它还是延迟问题。输入Token越多,模型处理时间越长。把Prompt和工具结果精简下来,既省钱又降延迟。
4. 一份能跑起来的服务化Agent骨架:FastAPI + LangGraph
4.1 目录结构与核心依赖
我最近给内部项目搭的Agent服务骨架是这么组织的:
agent-service/ ├── main.py ├── agent.py ├── tools.py └── requirements.txt依赖就四个核心库:
fastapi uvicorn langgraph langchain-openai其中langgraph负责状态图编排,langchain-openai负责模型接入,FastAPI只做最外层的HTTP接入。为了演示方便,我用了OpenAI兼容接口,实际你可以换成任何模型网关。
4.2 状态图编排:七个决策点落在代码里
先看tools.py,定义两个最普通的工具,一个是查订单状态,一个是查天气:
TOOL_MAP = {} def register_tool(name: str): def wrapper(func): TOOL_MAP[name] = func return func return wrapper @register_tool("query_order_status") def query_order_status(order_id: str) -> str: """模拟查询订单状态""" return f"订单{order_id}当前状态:已发货,预计明天送达。" @register_tool("query_weather") def query_weather(city: str) -> str: """模拟查询天气""" return f"{city}今日多云,18-24摄氏度。"然后是agent.py,这是核心。我用一个结构化输出让模型决定“要不要调工具、调哪个、参数是什么”,再用LangGraph把这几个节点串起来:
from typing import TypedDict, Literal from pydantic import BaseModel from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langgraph.graph import StateGraph, START, END from tools import TOOL_MAP SYSTEM_PROMPT = ( "你是一个客服Agent。请判断是否需要调用工具来回答用户问题。" "如果不需要工具,直接给出答案。" "注意:不要编造订单信息,必须用工具查询结果回答。" ) class AgentState(TypedDict): messages: list need_tool: bool tool_name: str | None tool_args: dict tool_result: str | None step: int class Plan(BaseModel): need_tool: bool tool_name: str | None = None tool_args: dict = {} reply: str | None = None # 第一个节点:意图与计划(对应决策点1、2、3) def plan_node(state: AgentState): llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) llm_planner = llm.with_structured_output(Plan) messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"] plan = llm_planner.invoke(messages) new_state = { "need_tool": plan.need_tool, "tool_name": plan.tool_name, "tool_args": plan.tool_args, "step": state["step"] + 1, } # 如果模型认为可以直接回答,就把回复追加进消息 if not plan.need_tool and plan.reply: new_state["messages"] = state["messages"] + [AIMessage(content=plan.reply)] return new_state # 第二个节点:执行工具(对应决策点3、4) def execute_node(state: AgentState): tool_name = state.get("tool_name") tool_args = state.get("tool_args") or {} if tool_name in TOOL_MAP: try: result = TOOL_MAP[tool_name](**tool_args) content = f"工具{tool_name}返回:{result}" except Exception as exc: content = f"工具{tool_name}调用失败:{str(exc)}" else: content = f"工具{tool_name}不存在" tool_message = AIMessage(content=content) return { "messages": state["messages"] + [tool_message], "tool_result": content, } # 第三个节点:生成最终回复(对应决策点5、6) def respond_node(state: AgentState): llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"] reply = llm.invoke(messages) return {"messages": state["messages"] + [AIMessage(content=reply.content)]}条件边用来表示决策点:
def after_plan(state: AgentState): # 决策点3:要不要调用工具 if state.get("need_tool"): return "execute" return "respond" def after_execute(state: AgentState): # 决策点5:信息是否足够,本轮是否继续计划 # 这里简单限制最多迭代3次,防止死循环 if state.get("step", 0) >= 3: return "respond" return "plan" def build_graph(): graph = StateGraph(AgentState) graph.add_node("plan", plan_node) graph.add_node("execute", execute_node) graph.add_node("respond", respond_node) graph.add_edge(START, "plan") graph.add_conditional_edges( "plan", after_plan, {"execute": "execute", "respond": "respond"}, ) graph.add_conditional_edges( "execute", after_execute, {"plan": "plan", "respond": "respond"}, ) graph.add_edge("respond", END) return graph.compile() agent_app = build_graph() def run_agent(session_id: str, user_message: str) -> str: initial_state = { "messages": [HumanMessage(content=user_message)], "need_tool": False, "tool_name": None, "tool_args": {}, "tool_result": None, "step": 0, } result = agent_app.invoke(initial_state) return result["messages"][-1].content这段代码对应的决策链路是:进入plan节点,模型判断要不要工具;如果要,进execute执行,执行完回到plan,模型看到工具结果后再决定是根据结果直接回答,还是继续调用下一个工具。step字段就是决策点7的兜底——最多跑3轮,超过就强制结束。
4.3 HTTP接口、任务队列与并发控制
main.py很简单:
from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel from agent import run_agent app = FastAPI() class ChatRequest(BaseModel): session_id: str message: str class ChatResponse(BaseModel): reply: str @app.post("/chat", response_model=ChatResponse) def chat(req: ChatRequest): reply = run_agent(req.session_id, req.message) return ChatResponse(reply=reply)注意,这个骨架只是演示短任务同步返回。真实项目里,我通常会把run_agent改成异步任务丢进队列,接口收到请求后立刻返回task_id,然后由worker执行并把结果写入Redis,客户端轮询/tasks/{task_id}拿结果。这样既避免长连接占用HTTP worker,也能方便地控制同时在途任务数量,避免把模型服务和工具服务打到超时。
如果你暂时不想引入Celery,可以用FastAPI的BackgroundTasks或者直接起一个asyncio.Queue消费端,都能接受。核心原则是:Agent长任务不要阻塞在请求线程里。
5. 从低代码平台到交易场景:几个热门方向的下水感受
5.1 低代码平台不是银弹
像扣子Coze这类可视化Agent平台,最近热度很高,我也用它搭过几个原型。说实话,做活动话术、私域客服、简单的信息查询助手,拖拽节点确实比写代码快得多。你可以在上面注册工具、编排节点、一键发布,甚至连知识库都可以直接挂上去。
但低代码平台的限制也很明显:深度定制的权限模型、复杂的内部系统对接、细粒度的可观测性,这些都不好做。我通常的建议是:用低代码平台验证业务逻辑,验证通过后如果要对公网提供服务、对接核心系统,再迁移到代码工程里。平台帮你快速看到了效果,但不要把核心链路绑死在无法审计的工具上。
5.2 自动发消息和交易Agent的坑
最近总能刷到“用AI Agent让小红书自动发消息”这类教程。从纯技术角度看,调用接口、定时脚本、模拟人工操作都不难,但难的是平台风控和内容合规。频繁自动发消息很容易触发限流甚至封号,更严重的是如果脚本失控,会制造大量垃圾内容。我的态度是:这种自动化的前提必须是可控——频率限制必须有、内容审核必须有、人工退出开关必须有。不要为了跑一个demo,把账号和口碑搭进去。
期货交易Agent也是热度很高的话题。技术上,让LLM分析新闻、生成信号、甚至组装下单指令都能做到,但实盘交易不是“调用模型下单”这么简单。行情是毫秒级的,模型推理是秒级的;交易链路必须拆成“信号层”和“执行层”,信号层可以用LLM辅助分析,执行层的风控、限价、撤单、最大回撤都必须是确定性代码。还有一个更现实的问题:普通个人是否具备实盘交易资质、是否了解杠杆风险。把LLM接到实盘下单这个事,风险远大于收益,我强烈不建议在没有合规和专业风控的情况下尝试。
5.3 阿里云Agent白皮书这类资料怎么读才值
阿里云出过AI Agent白皮书,我也完整读过。这种东西的价值不在于给你现成的代码,而在于帮你建立行业全景图:Agent当前有哪些主流架构、企业落地需要哪几层能力、评估指标是什么。读的时候别只盯着架构图,要结合自己的业务问几个问题:我现在是做单Agent还是多Agent?我的Agent需要哪些工具?谁为Agent的错误负责?
带着这些问题去读,白皮书才能变成决策参考,而不是收藏夹里的又一个“等有时间再看”。
最后聊避坑清单,这些都是我踩过的:
- 不要在一条System Prompt里编排全部逻辑。工具一多,Prompt必乱,尽早拆到代码节点里。
- 不要一上来就上多Agent。大多数业务,一个Agent配几个工具就够用了。多Agent带来的协调、上下文传递、调试复杂度是翻倍增长的。
- 别拿Rust重写一切。除非你要做高吞吐的Agent网关或模型接入底座,不然业务Agent用团队最熟的语言写就好。
- 别忽略可观测性。每个节点要有trace、耗时、token计数和决策日志,否则线上出事只能靠猜。
- Agent每一步都要有退出阀。最大迭代次数、超时、人工确认,缺一不可。
我自己在实际操作中的体会是,Agent工程化做得越久,越明白“智能”其实是被一层层决策点框出来的:模型只在有限的岔路口做选择,其余路径都由代码保证。把七要素当成需求清单,把七个决策点当成状态机里的条件分支,Agent的工程实现就能从“看起来智能”变成“可以被测试、被运维、被信任”。希望这份骨架和心得,能让你少走几步弯路。