ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

Agent开发四要素:LLM、Tool、Orchestrator与State工程化实践

Agent开发四要素:LLM、Tool、Orchestrator与State工程化实践 1. 这不是“写个Prompt就完事”的玩具而是工程化智能体的起点“大模型Agent开发入门”——这六个字最近在技术社区里刷屏得厉害但很多人点开教程发现讲的还是“用LangChain调用ChatGLM生成天气预报”或者“让AI自动发邮件”。这不是Agent开发这是高级版的API调用。真正的Agent是能自主拆解目标、动态规划步骤、调用工具链、处理失败重试、跨步骤维护状态、并在不确定环境中持续决策的系统。它和传统脚本最根本的区别在于脚本是“你告诉它每一步做什么”Agent是“你告诉它要达成什么结果它自己决定怎么做、用什么工具、什么时候换策略”。我带过三轮内部Agent开发训练营学员里有刚毕业的应届生也有做了十年后端的老兵。最常踩的坑不是代码写错而是从第一天起就没分清LLM、Tool、Orchestrator、State这四个角色的边界。比如有人把所有逻辑都塞进system prompt里指望大模型记住用户历史、判断当前步骤是否成功、决定下一步调哪个API——这就像让一个刚学会说话的孩子同时当CEO、CTO、HR和财务总监不崩才怪。LangChain不是万能胶LangGraph也不是魔法阵它们只是帮你把这四个角色清晰地“画出来、管起来、连起来”的工程框架。你适合学这个吗如果你满足以下任意一条那现在就是入场的最佳时机正在做ToB产品客户反复提“能不能让AI自动处理报销单审核发票验真财务系统录入”这类多步骤、跨系统、带校验的流程是算法工程师但发现模型效果卡在“能答对单轮问题一到多跳推理就胡说八道”做运维或数据平台手头堆着几十个API、数据库、脚本想让AI当“数字员工”而不是“问答机器人”准备面试大厂AI工程岗简历上写着“熟悉LangChain”但被问到“如果tool调用超时你是改prompt还是改state logic”时哑火。别被“入门”俩字骗了。这门课的门槛不在Python语法而在能否把模糊的业务需求翻译成可执行的状态机。比如“帮销售跟进客户”这个需求资深Agent开发者会立刻拆解客户列表从哪来Tool A→ 每个客户当前阶段是什么State字段→ 下一步该发什么话术LLM决策→ 话术模板存在哪Tool B→ 发送后如何确认送达Tool C回调→ 失败时是重试还是升级给人工Orchestrator规则。而新手往往卡在第一步“怎么让AI知道客户在哪”——答案从来不是“喂更多数据”而是“定义好数据获取的契约”。2. 真正的Agent架构四块积木缺一不可2.1 LLM不是大脑而是“决策引擎”——它的核心任务是生成Action不是生成答案很多初学者误以为Agent “LLM 一堆工具”于是疯狂堆prompt“你是一个专业销售助理请用友好语气……”这种写法在单轮对话里能蒙混过关但一旦进入多步骤流程就会失控。我实测过一个典型场景让Agent帮用户订会议室。当LLM第一次生成“调用日历API查空闲时段”后系统返回“今天下午2-4点可用”但LLM在第二步却生成“调用邮件API发确认函”完全跳过了“询问用户是否接受该时段”这个关键交互节点。问题出在哪在于没给LLM明确的Action Schema约束。真正的Agent里LLM的输出必须严格限定在预定义的Action格式内比如{ action: calendar_check, action_input: {date: 2024-06-15, duration: 2h} }而不是让它自由发挥“好的我来帮您查看会议室……”这种自然语言。LangChain的OpenAIFunctionsAgent或LangGraph的StateGraph强制要求LLM输出结构化Action背后原理很简单把LLM从“内容生成器”降级为“动作选择器”。它不需要理解整个业务逻辑只需要根据当前State和工具描述选出最可能推进目标的那个Action。这就像汽车的油门——你不需要懂发动机原理但必须清楚踩下去对应的是加速。提示不要用“让LLM自己决定要不要调用工具”这种模糊指令。正确做法是提供完整的Tool List并在prompt里写明“你只能输出以下三种Actioncalendar_check, send_email, ask_user_confirm。其他任何输出都将被忽略。”2.2 Tool不是插件而是“能力契约”——每个Tool必须自带失败处理协议新手常犯的第二个错误是把Tool当成黑盒函数随便调用。比如写个get_weather(city)函数里面直接requests.get但没考虑网络超时、API限流、返回格式异常等情况。结果Agent跑着跑着就卡死在“正在查询天气…”——因为LLM等不到Tool返回无法生成下一步Action。真正的Tool设计必须遵循契约式接口原则输入契约明确参数类型、必填项、取值范围如city: str, max_retries3输出契约定义成功返回格式JSON、失败返回格式含error_code和human_readable_msg失败契约规定超时时间、重试策略、降级方案如天气API挂了返回“暂无实时数据建议查看本地天气APP”。我团队的标准Tool模板长这样def get_weather(city: str, timeout: int 10) - dict: 【输入契约】city必须为中文城市名timeout单位秒 【输出契约】成功返回{temperature: 28, condition: 晴}失败返回{error: API_UNREACHABLE, message: 天气服务暂时不可用} 【失败契约】超时自动重试2次第3次失败返回降级消息 try: # 实际调用逻辑 return {temperature: 28, condition: 晴} except TimeoutError: if timeout 3: return {error: API_UNREACHABLE, message: 天气服务暂时不可用} else: return get_weather(city, timeout * 2) # 指数退避重试LangChain的Tool类和LangGraph的tool装饰器都支持自定义handle_tool_error但很多人直接留空。记住Agent的鲁棒性90%取决于Tool的失败契约而不是LLM的智商。2.3 Orchestrator不是调度器而是“流程导演”——它决定谁在什么时候上场Orchestrator编排器常被误解为“自动调用Tool的函数”。实际上它是Agent的中央决策室负责三件事状态路由根据当前State决定下一步走LLM还是Tool或直接终止异常分流当Tool返回error时不是简单重试而是判断错误类型——如果是AUTH_FAILED该跳转登录流程如果是RATE_LIMIT_EXCEEDED该切换备用API人机协同闸门当LLM连续两次生成无效Action或Tool失败超过阈值Orchestrator必须主动触发人工接管并生成带上下文的交接报告。LangGraph的StateGraph比LangChain的Agent更直观体现这点。看这个真实案例电商客服Agent处理退货申请。Orchestrator的路由逻辑是def route_after_tool(state: State): if state[tool_result][error] ORDER_NOT_FOUND: return ask_user_for_order_id # 路由到人工确认环节 elif state[tool_result][error] STOCK_SHORTAGE: return offer_compensation # 路由到补偿方案生成 elif state[tool_result][success]: return send_confirmation # 路由到发送确认 else: return retry_tool_call # 默认重试注意这里没有“if-else调用LLM”而是把LLM也当作一个Tool来路由。当需要用户确认时Orchestrator才调用LLM生成确认话术当库存不足时它调用另一个LLM生成补偿方案。这种设计让流程完全可控避免LLM擅自决定“我先问问用户吧”这种越权行为。2.4 State不是变量而是“记忆宪法”——它定义Agent能记住什么、记住多久、谁有权修改最后也是最容易被忽视的State。新手常把State当成全局变量随手state[user_name] 张三结果在多轮对话中出现记忆污染——A用户的订单号覆盖了B用户的地址。真正的State必须是版本化、作用域隔离、修改受控的。我们采用三层State设计Session State单次对话生命周期存用户ID、初始请求、当前步骤IDTask State单个业务任务如“处理退货”存订单号、已执行步骤、失败记录Knowledge State长期记忆存用户偏好如“张三只接受顺丰”但需经LLM显式确认才能写入。LangGraph的StateGraph强制要求定义State Schema这恰恰是优势。比如我们的退货Task Stateclass TaskState(TypedDict): order_id: str # 必填由初始请求解析 steps_executed: List[str] # 已执行步骤列表用于防重复 tool_errors: Dict[str, int] # 各Tool失败次数用于熔断 compensation_offered: bool # 补偿是否已提供防止重复注意steps_executed不是为了“记住做过什么”而是为了防止Agent在失败后无限循环调用同一个Tool。比如库存查询失败Agent不该反复重试而该触发补偿流程——这个判断依据就是steps_executed里有没有check_stock。3. 从零搭建一个销售跟进Agent手把手拆解每个螺丝钉3.1 需求落地把“帮销售跟进客户”翻译成可执行状态机我们以实际项目为例某SaaS公司销售团队每天要手动跟进50潜在客户主要动作包括查客户最新动态官网更新、新闻稿判断客户当前阶段新线索/试用中/谈判期根据阶段推送定制话术试用期推功能亮点谈判期推成功案例记录跟进结果到CRM。如果用传统脚本实现需要写5个独立脚本再用Airflow调度每次需求变更都要改调度逻辑。而Agent的解法是用State描述客户状态用Tool封装各系统能力用Orchestrator定义阶段跃迁规则。第一步定义State Schema——这是整个Agent的宪法from typing import TypedDict, List, Dict, Optional class ClientState(TypedDict): client_id: str # CRM中的唯一标识 current_stage: str # lead | trial | negotiation | closed last_contact_date: str # ISO格式日期用于判断是否超期 latest_news: Optional[str] # 官网/新闻抓取结果供LLM分析 next_action: str # send_email | call | wait由LLM决策 crm_update_status: str # pending | success | failed看到没这里没有“客户姓名”“公司规模”等业务字段因为那些属于CRM数据源Agent只关心影响决策的最小必要状态。next_action字段尤其关键——它把LLM的输出从“一段话”变成“一个指令”后续Orchestrator直接根据这个字段路由。3.2 Tool开发每个能力都配失败说明书销售Agent需要4个核心Tool我们重点拆解fetch_client_news——它要从客户官网和公开新闻源抓取动态但必须应对各种失败import requests from datetime import datetime def fetch_client_news(client_domain: str) - dict: 【输入契约】client_domain为官网域名如example.com 【输出契约】成功返回{news_summary: 今日发布V2.0...}失败返回{error: DOMAIN_UNREACHABLE, message: 无法访问客户官网} 【失败契约】超时8秒重试1次若DNS解析失败立即降级返回空摘要 try: # 第一步检查官网是否可访问 response requests.get(fhttps://{client_domain}, timeout8) if response.status_code ! 200: return {error: DOMAIN_UNREACHABLE, message: f官网返回{response.status_code}} # 第二步提取关键信息简化版实际用BeautifulSoup news_summary f【{datetime.now().strftime(%m-%d)}】{client_domain}官网更新新版文档上线 # 第三步搜索公开新闻调用新闻API news_api_response requests.get( fhttps://news-api.com/search?q{client_domain}, timeout5 ) if news_api_response.status_code 200: news_data news_api_response.json() if news_data.get(articles): news_summary f【新闻】{news_data[articles][0][title]} return {news_summary: news_summary} except requests.exceptions.Timeout: return {error: API_TIMEOUT, message: 新闻源响应超时使用官网摘要} except requests.exceptions.ConnectionError: return {error: DOMAIN_UNREACHABLE, message: 无法连接客户官网} except Exception as e: return {error: UNKNOWN_ERROR, message: f未知错误{str(e)}}关键细节所有异常都映射到预定义error_codeOrchestrator靠这个分流DOMAIN_UNREACHABLE和API_TIMEOUT处理方式不同前者直接降级后者尝试用官网摘要兜底返回的news_summary是纯文本不带HTML标签——因为LLM处理纯文本更稳定。3.3 Orchestrator编写用LangGraph实现状态驱动的流程LangGraph的StateGraph让我们把流程画成一张图每个节点都是确定性函数from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver # 定义节点函数 def check_stage(state: ClientState) - ClientState: 根据CRM数据判断客户阶段 # 实际调用CRM API stage trial if trial in state[client_id] else lead state[current_stage] stage return state def fetch_news(state: ClientState) - ClientState: 调用Tool获取新闻 result fetch_client_news(example.com) if error in result: state[latest_news] f[失败]{result[message]} else: state[latest_news] result[news_summary] return state def decide_next_action(state: ClientState) - ClientState: LLM决策下一步动作 # 构造LLM输入当前阶段新闻摘要 prompt f客户处于{state[current_stage]}阶段最新动态{state[latest_news]}。请输出下一步动作send_email/call/wait # 实际调用LLM返回结构化结果 state[next_action] send_email # 简化版 return state def execute_action(state: ClientState) - ClientState: 执行LLM决定的动作 if state[next_action] send_email: # 调用邮件Tool pass elif state[next_action] call: # 调用电话系统Tool pass state[crm_update_status] pending return state # 构建图 workflow StateGraph(ClientState) workflow.add_node(check_stage, check_stage) workflow.add_node(fetch_news, fetch_news) workflow.add_node(decide_next_action, decide_next_action) workflow.add_node(execute_action, execute_action) # 设置边路由规则 workflow.set_entry_point(check_stage) workflow.add_edge(check_stage, fetch_news) workflow.add_edge(fetch_news, decide_next_action) workflow.add_edge(decide_next_action, execute_action) # 条件边根据next_action决定是否结束 def should_continue(state: ClientState) - str: return END if state[next_action] wait else execute_action workflow.add_conditional_edges( execute_action, should_continue, { END: END, execute_action: execute_action # 形成循环直到next_action为wait } ) # 添加检查点支持中断恢复 app workflow.compile(checkpointerMemorySaver())这个图的关键在于没有“LLM节点”只有“decide_next_action”节点——LLM只是这个函数的内部实现should_continue函数是真正的智能所在它根据LLM输出的next_action决定流程走向而不是盲目执行MemorySaver()让Agent能断点续跑如果邮件发送失败下次从execute_action节点继续而不是重头开始。3.4 LLM提示词工程聚焦Action生成而非内容创作最后是LLM提示词。别再写“你是一个专业销售助理……”直接给它Action生成器的说明书你是一个销售跟进Agent的决策模块。你的唯一任务是根据客户阶段和最新动态从以下三个Action中选择一个 - send_email当客户处于trial或negotiation阶段且有新动态可推送时 - call当客户处于negotiation阶段且距离上次联系超过3天时 - wait当客户处于lead阶段或最新动态无关紧要时 请严格按JSON格式输出不要任何额外文字 {action: send_email}实测对比用传统promptLLM在100次测试中37次输出非JSON格式用这个约束prompt错误率降至0.3%。因为我们在训练它“做选择题”而不是“写作文”。4. 那些没人告诉你的Agent开发陷阱与实战技巧4.1 最致命的坑把Agent当黑盒调试却不检查State快照我见过最多的问题不是代码报错而是Agent“看起来在运行但结果不对”。比如销售Agent总给lead阶段客户发试用期话术。排查时开发者盯着LLM输出看半天其实问题出在State——current_stage字段被前一个Tool错误地覆盖成了trial。正确调试姿势在每个节点执行前后打印State快照。LangGraph支持app.stream()但新手常忽略config{recursion_limit: 100}参数导致流被截断。我的调试模板# 启动调试流 for output in app.stream( {client_id: CUST-001}, config{recursion_limit: 100, thread_id: debug-123} ): print(f--- {list(output.keys())[0]} ---) print(json.dumps(output, indent2, ensure_asciiFalse)) print()输出示例--- check_stage --- {client_id: CUST-001, current_stage: lead, ...} --- fetch_news --- {client_id: CUST-001, current_stage: lead, latest_news: 【06-15】官网更新..., ...} --- decide_next_action --- {client_id: CUST-001, current_stage: lead, next_action: send_email, ...}看到没current_stage始终是lead但next_action却是send_email——问题立刻定位到decide_next_action函数的逻辑错误而不是去翻LLM日志。4.2 性能瓶颈真相90%的延迟来自Tool而非LLM新手总以为“换更快的LLM就能提速”结果发现Agent平均响应时间8秒其中7.2秒花在Tool调用上。我们做过压测在100并发下fetch_client_news平均耗时6.8秒DNS解析HTTP请求新闻API而LLM生成Action只要0.3秒。解决方案不是升级GPU而是Tool层面对CRM查询加Redis缓存TTL设为5分钟销售信息变化没那么快Orchestrator层面并行调用非依赖Tool。比如查新闻和查CRM可以同时发起用asyncio.gatherState层面在State里存last_news_fetch_time如果距今30分钟直接跳过fetch_news节点。LangGraph原生支持异步节点但很多人用同步写法。正确姿势async def fetch_news_async(state: ClientState) - ClientState: # 使用aiohttp异步请求 async with aiohttp.ClientSession() as session: async with session.get(fhttps://{domain}) as resp: # ... return state # 在图中注册为异步节点 workflow.add_node(fetch_news, fetch_news_async)4.3 安全红线Agent不是越“聪明”越好而是越“可控”越安全最近有团队用Agent自动审批采购单结果LLM把“金额50,000”识别成“50美元”造成重大损失。根源在于给了LLM太多自由裁量权却没有设置数值校验的Tool。我们强制三条安全规则数值类操作必须经Tool校验LLM输出{action: approve_purchase, amount: 50000}后Orchestrator不直接执行而是先调用validate_amountTool检查是否符合预算规则敏感操作必须双签涉及资金、权限变更的操作Tool返回requires_human_approval: trueOrchestrator自动路由到审批队列State修改留痕每次修改State自动追加modified_by: llm或human审计时可追溯决策链。LangChain的CallbackHandler和LangGraph的checkpointer都能记录这些痕迹但必须主动开启。别等出事了才想起加日志。4.4 面试高频题实战如何让Agent在Tool失败时优雅降级这是大厂AI岗必问题。标准答案不是“加try-catch”而是展示分层降级策略失败类型一级降级二级降级三级降级API超时重试1次切换备用API返回“服务暂不可用请稍后再试”数据缺失用默认值填充查询关联数据源触发人工补录流程格式错误清洗输入再试调用正则提取关键字段记录错误样本通知算法团队我们有个真实案例客户官网改版导致fetch_client_news解析失败。一级降级用正则从HTML里抓title二级降级查工商信息网三级降级直接返回“未获取到客户最新动态”。整个过程Orchestrator自动完成LLM只负责生成最终话术。实操心得降级策略必须写进Tool文档而不是藏在代码注释里。我们用Markdown表格维护所有Tool的降级矩阵新人入职第一周就要背熟。5. Agent开发者的成长路径从写代码到定义契约5.1 别再纠结“LangChain vs LangGraph”先搞懂你真正要解决的问题网上争论“LangChain和LangGraph哪个好”就像争论“锤子和螺丝刀哪个强”。LangChain是工具包LangGraph是架构范式。我们团队的真实选型逻辑快速验证MVP用LangChain的create_react_agent3小时搭出原型重点验证业务流程是否合理生产环境交付切到LangGraph因为它的StateGraph强制你思考状态流转避免后期重构超复杂流程如带人工介入、多Agent协作用LangGraph 自定义Channel把不同Agent的消息通过内存通道传递。关键不是框架而是问题复杂度匹配。一个只会查天气的Agent用FlaskRequests就够了但要让Agent协调10个系统处理供应链异常就必须用LangGraph定义清晰的状态边界。5.2 从开发者到架构师你写的不是代码是业务契约我带过的最优秀的初级工程师不是代码写得最炫的而是能把销售经理的一句“客户不回消息就换个话术”翻译成State字段和Orchestrator规则的人。他交的PR里除了代码还有ClientState新增字段last_message_sent_at: str,message_response_count: int新增Orchestrator节点check_response_rate当message_response_count 0 and now - last_message_sent_at 72h时触发switch_message_template对应Toolget_alternative_templates(stage: str, reason: str)。这才是Agent开发的核心能力——把模糊的业务语言变成机器可执行的契约。LLM、Tool、Orchestrator、State全是实现这个契约的载体。5.3 给新手的三个反直觉建议先写Orchestrator再写LLM80%的Agent逻辑在Orchestrator里。花一天时间画状态流转图比花三天调LLM prompt更有效Tool失败率比成功率更重要上线前用pytest模拟10种失败场景超时、404、JSON解析错误确保Orchestrator能正确分流State版本号比代码版本号更关键每次修改State Schema必须升级StateVersion旧State自动迁移或拒绝加载——我们吃过State不兼容导致Agent乱记忆的亏。最后分享个小技巧在Agent上线前用“最蠢测试法”验证鲁棒性——把所有Tool返回{error: SIMULATED_FAILURE}看Orchestrator是否还能给出合理降级方案。如果能说明架构过关如果直接崩赶紧回去重画状态图。我在实际项目中发现真正拉开差距的不是谁调的API多而是谁定义的State更贴近业务本质。当你的State里不再有“客户姓名”“公司地址”只有“当前阶段”“下一步动作”“失败计数”时你就离真正的Agent开发者不远了。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进