☰
AI Agent工程化实战:从七要素到状态机决策点
2026/10/7 12:58:26 网站建设 项目流程

这几年只要聊到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为例,每轮执行中至少会出现下面七个选择题:

  1. 意图是否清晰:用户说“我想退货”,但没给订单号,这时候是直接查还是先追问?
  2. 是否需要记忆或外部知识:这个问题能不能靠当前对话回答?要不要检索用户历史订单、商品知识库?
  3. 该调用哪个工具、传什么参数:决定工具名称和入参,这一步通常是模型负责的。
  4. 工具返回值是否合法:订单API返回了404,是订单号错了还是服务异常?要不要换个工具或重试?
  5. 信息是否足够结束:拿到工具结果后,能不能生成最终答复?还是需要再追问、再查一个工具?
  6. 输出是否需要人工确认:涉及退款、发消息、交易操作时,是不是要先走人工审批?
  7. 失败与超时如何兜底:模型超时、工具超时、连续失败,有没有降级话术或转人工通道?

这七个决策点不一定每轮都出现,但它们必须被显式设计在代码里。比如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极快,适合运营自建话术机器人,但遇到复杂权限、私有化部署、深度定制时天花板明显。
维度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 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的工程实现就能从“看起来智能”变成“可以被测试、被运维、被信任”。希望这份骨架和心得,能让你少走几步弯路。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询