这几年只要聊到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 / LangGraphPython团队上手快状态图编排直观适合快速验证和复杂流程编排。缺点是你得接受它迭代快、API变动频繁带来的“追版本”成本。Spring AIJava/Spring团队集成成本低企业级基础设施很全配置、监控、AOP但生态和社区资料目前比LangChain少一些。Rust 自研状态机并发模型确实强适合做高吞吐的Agent网关、统一模型接入层这类基础设施。但如果你只是想快速把业务跑起来用Rust从零搭Agent会痛到你怀疑人生。低代码平台类似扣子Coze这类拖拽节点做demo极快适合运营自建话术机器人但遇到复杂权限、私有化部署、深度定制时天花板明显。维度LangGraphSpring AIRust自研低代码平台状态编排强中自研成本高可视化但灵活度低团队门槛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 tokenAgent平均跑5轮一次任务光输入Token就是(800 2000 1500) × 5 21500 token这还没算生成输出。如果一天跑几千个任务成本就非常可观。我从成本角度给的优化建议是模型分级规划、推理、总结用强模型意图分类、实体提取、格式化输出这类重复任务用轻量模型甚至可以本地搞定。不要无脑把工具结果塞进上下文工具返回一大段JSON先抽取出关键字段再喂给模型剩下的一律丢弃。消息压缩超过一定轮数后把旧消息摘要成一段话而不是无限追加。命中缓存相同或相似的上下文请求直接用缓存结果尤其是FAQ类任务能省下一大笔重复调用的钱。Token不只是成本问题它还是延迟问题。输入Token越多模型处理时间越长。把Prompt和工具结果精简下来既省钱又降延迟。4. 一份能跑起来的服务化Agent骨架FastAPI LangGraph4.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(modelgpt-4o-mini, temperature0) llm_planner llm.with_structured_output(Plan) messages [SystemMessage(contentSYSTEM_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(contentplan.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(contentcontent) return { messages: state[messages] [tool_message], tool_result: content, } # 第三个节点生成最终回复对应决策点5、6 def respond_node(state: AgentState): llm ChatOpenAI(modelgpt-4o-mini, temperature0) messages [SystemMessage(contentSYSTEM_PROMPT)] state[messages] reply llm.invoke(messages) return {messages: state[messages] [AIMessage(contentreply.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(contentuser_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_modelChatResponse) def chat(req: ChatRequest): reply run_agent(req.session_id, req.message) return ChatResponse(replyreply)注意这个骨架只是演示短任务同步返回。真实项目里我通常会把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的工程实现就能从“看起来智能”变成“可以被测试、被运维、被信任”。希望这份骨架和心得能让你少走几步弯路。