基于LLM多Agent系统的ERP流程自动化：架构设计与工程实践

1. 项目概述与核心思路上次我们聊了从ERP系统出发设计LLM多Agent系统的整体架构和核心思想算是把“为什么做”和“大致怎么做”讲清楚了。今天这篇咱们就深入到“具体怎么做”的层面把骨架填上血肉。很多朋友在后台留言说对Agent之间的协作机制、状态管理以及如何让它们真正理解ERP业务逻辑特别感兴趣这正是本篇要解决的核心问题。简单来说一个能处理复杂ERP流程的多Agent系统绝不是简单地把几个大语言模型LLM调用接口拼在一起。它更像是在组建一个高度专业化的虚拟项目团队每个Agent都是这个团队里的一位专家有明确的职责、专属的工具箱以及一套严谨的协作沟通协议。我们的目标是让这个“虚拟团队”能够自主、可靠地处理从销售订单录入到财务凭证生成这样一条完整的业务链。这背后涉及到Agent的“大脑”推理与决策、“手”工具调用和“嘴”沟通协作如何协同工作。我会结合我们团队在落地过程中的实际案例拆解设计细节、分享踩过的坑以及那些在标准文档里不会写的调试心得。2. 核心Agent的角色定义与能力构建设计多Agent系统的第一步也是最重要的一步就是明确每个Agent的“人设”。你不能笼统地定义一个“业务处理Agent”这就像招人时只说“我们需要一个能干活的”结果必然是一团糟。在ERP场景下我们需要的是高度专业化的角色。2.1 基于业务域划分核心Agent角色我们的设计是基于经典的企业业务流拆解出以下几个核心Agent角色销售订单处理Agent这是流程的起点。它的核心职责是解析用户输入的订单需求可能是自然语言描述也可能是半结构化数据并提取关键实体如客户信息、产品SKU、数量、价格条款、交货日期等。它需要具备强大的信息抽取和标准化能力。库存与供应链协调Agent订单确认后立即需要它来介入。它的工作是查询实时库存、检查在途物料、评估生产能力。它需要访问库存数据库、MRP物料需求计划系统等并做出“有货直发”、“安排生产”或“提示缺料”的判断。生产计划Agent如果涉及对于需要生产的订单这个Agent负责将销售订单转化为具体的生产工单考虑BOM物料清单、工艺路线、设备负荷和交货期倒排。财务合规与定价Agent这个Agent负责所有与钱相关的规则。它需要应用复杂的定价策略客户等级折扣、促销活动、计算税费、审核信用额度并确保整个订单符合公司的财务政策。工作流编排与状态管理Agent或称“协调者Agent”这是整个系统的“项目经理”。它不直接处理具体业务而是监督流程的推进根据其他Agent的反馈决定下一步该激活哪个Agent并维护一个全局的“订单上下文状态”。注意角色的粒度需要权衡。角色太少单个Agent过于复杂容易出错角色太多则通信开销巨大协调困难。我们的经验是初期可以按照企业内现有的部门职能或关键系统模块来划分这样业务逻辑最清晰。2.2 为每个Agent装备专属“工具链”一个只有“大脑”LLM的Agent是纸上谈兵。它必须能操作现实系统这就需要“工具”。我们为每个Agent设计了一套工具函数Tools这些函数是对后端ERP API、数据库查询、规则引擎等能力的封装。例如对于库存与供应链协调Agent它的工具链可能包括check_inventory(sku, warehouse_id): 查询特定仓库的实时库存。check_supply_plan(sku, required_date): 查询该物料的未来供应计划采购在途、生产计划。reserve_inventory(order_id, sku, quantity): 为指定订单预留库存。calculate_lead_time(sku, quantity): 计算生产或采购的预估提前期。这些工具的定义需要非常精确包括函数名、描述、输入参数类型、说明和返回值的结构。LLM如GPT-4能够根据清晰的工具描述在合适的时机选择并调用正确的工具。实操心得工具描述的“咒语”工程工具的描述description至关重要它直接决定了LLM是否能正确理解和使用该工具。不要只写“查询库存”而应该写成“根据产品SKU和仓库编号查询该仓库下的可用现货数量在库量减去已预留量。如果仓库编号为空则返回该SKU在所有仓库的总可用量。” 这种描述方式包含了意图、参数逻辑和业务规则。2.3 Agent的“大脑”定制提示词工程与少样本学习虽然底层可能使用同一个大模型但每个Agent需要有专属的“思维模式”这是通过**系统提示词System Prompt和少样本示例Few-shot Examples**来实现的。销售订单处理Agent的系统提示词会强调 “你是专业的销售订单分析员。你的任务是从用户的对话或文本中精确提取结构化订单信息。你必须关注以下字段客户名称、产品代码与数量、单价、总价、交货日期、特殊条款。对于模糊信息你必须主动询问澄清而不是猜测。输出必须为标准的JSON格式。”然后我们会提供几个少样本示例展示一段模糊的用户输入和期望的、经过澄清后的JSON输出。这相当于给Agent做了上岗培训。财务合规与定价Agent的提示词则完全不同会强调 “你是严格的财务合规审核员。你必须依据最新版的《公司定价与信用政策手册》来审核订单。重点关注客户信用等级对应的折扣上限、当前促销活动是否适用、税费计算规则如含税/不含税。任何违反政策的情况你必须明确拒绝并引用具体条款。”通过这种方式我们让同一个LLM底层能力在不同的提示词和示例“熏陶”下扮演了截然不同的专业角色。3. 多Agent协作机制的设计与实现角色定义好了如何让它们高效协作是关键。我们摒弃了简单的线性流水线Agent A做完传给Agent B因为ERP业务流充满分支和回环。例如财务审核可能否决订单需要退回销售修改。3.1 基于“发布-订阅”与“工作流引擎”的混合模式我们采用了一种混合架构核心协调者工作流编排与状态管理Agent充当核心协调者。它持有当前处理业务对象如订单的完整上下文状态机。消息总线我们引入了一个轻量级的内部消息总线或事件驱动架构。当某个Agent完成一项任务或需要触发下一环节时它会向总线“发布”一个结构化事件。事件驱动协作协调者Agent“订阅”所有关键事件。它根据当前状态和接收到的事件决定下一步的流向并“通知”相应的Agent开始工作。例如事件SalesOrderExtracted销售订单处理Agent发布包含提取的订单JSON。协调者收到后更新状态为“待核库存”并通知库存与供应链协调Agent开始工作。库存Agent工作后发布事件InventoryChecked结果可能是{status: “sufficient”, reserved_id: “xxx”}或{status: “insufficient”, alternatives: […]}。协调者根据结果决定是通知财务Agent还是发布一个RequireHumanIntervention事件库存不足需人工确认。这种模式解耦了Agent之间的直接依赖使系统更灵活更容易扩展新的Agent或修改流程。3.2 共享上下文与状态管理在整个流程中订单数据上下文在不断丰富和演变。我们设计了一个共享上下文对象它随着流程在Agent间传递。这个对象不仅仅是原始数据还包括流程的元数据。{ “process_id”: “ORDER-20240520-001”, “current_stage”: “FINANCE_APPROVAL”, “created_at”: “2024-05-20T10:00:00Z”, “context_data”: { “customer”: {“id”: “C1001”, “name”: “XX公司”, “credit_level”: “A”}, “order_lines”: […], “inventory_reservation_id”: “RES-20240520001”, “pricing_summary”: {…}, “audit_trail”: [ // 审计轨迹记录每个Agent的操作和结果 {“agent”: “SalesAgent”, “action”: “extract”, “result”: “success”, “timestamp”: “…”}, {“agent”: “InventoryAgent”, “action”: “check_and_reserve”, “result”: “reserved”, “timestamp”: “…”} ] }, “errors”: [], “requires_human_input”: false }协调者Agent负责维护和更新这个上下文对象。每个Agent在执行时会收到与它相关的上下文切片执行完毕后再将结果写回。审计轨迹对于调试和追溯决策过程至关重要。3.3 Agent间通信协议标准化与容错Agent之间不能靠“自由对话”来协作必须定义严格的通信协议。我们采用了基于JSON的标准化消息格式{ “message_id”: “msg_001”, “from_agent”: “InventoryCoordinator”, “to_agent”: “WorkflowOrchestrator”, “message_type”: “TASK_RESULT”, // 或 TASK_REQUEST, ERROR, NOTIFICATION “content”: { “task_id”: “task_inv_check_001”, “status”: “COMPLETED”, “data”: {…}, // 任务产出数据 “error”: null }, “timestamp”: “…” }容错设计超时与重试每个任务都有超时设置。如果Agent在规定时间内未返回结果协调者会标记任务为超时可以根据策略重试或转人工。一致性检查关键步骤后设计一个“验证Agent”或由后续Agent对前序结果做简单逻辑校验。例如财务Agent在计算总价时会复核数量x单价是否与销售Agent提取的数据一致。降级策略当某个Agent如复杂的生产排程Agent反复失败时系统可以降级为发布一个“人工处理”事件并记录下已完成的步骤和失败原因保证流程不彻底中断。4. 系统核心组件的技术实现细节讲清楚了设计我们来看看一些关键部分如何用代码实现。这里不会贴出全部代码但会展示核心模式和思路。4.1 Agent基类与工具调用框架我们为所有Agent实现了一个基类它封装了与LLM的交互、工具调用和消息格式化。class BaseAgent: def __init__(self, name, system_prompt, tools, llm_client): self.name name self.system_prompt system_prompt self.tools {tool.name: tool for tool in tools} # 工具字典 self.llm llm_client def _parse_llm_response(self, response): # 解析LLM返回的JSON其中可能包含工具调用请求 # 格式如{“thought”: “…”, “action”: {“name”: “tool_name”, “args”: {...}}} pass def _execute_tool(self, tool_name, args): # 查找并执行工具 if tool_name in self.tools: return self.tools[tool_name].function(**args) else: raise ValueError(f“Tool {tool_name} not found.”) async def run(self, context): # 运行Agent的主循环 messages [ {“role”: “system”, “content”: self.system_prompt}, {“role”: “user”, “content”: self._format_task_for_llm(context)} ] while True: llm_response await self.llm.chat_completion(messages, …) parsed self._parse_llm_response(llm_response) if parsed[“action”] is None: # LLM认为任务完成返回最终结果 return parsed[“final_answer”] else: # 执行工具调用 tool_result self._execute_tool(parsed[“action”][“name”], parsed[“action”][“args”]) # 将工具执行结果作为新的上下文信息附加给LLM让它继续思考 messages.append({“role”: “assistant”, “content”: llm_response}) messages.append({“role”: “user”, “content”: f“Tool {parsed[‘action’][‘name’]} returned: {tool_result}”})4.2 工作流协调者的状态机实现协调者Agent的核心是一个状态机。我们使用了Python的transitions库来清晰定义状态和转移条件。from transitions import Machine class OrderWorkflow: states [‘idle’, ‘order_extracting’, ‘inventory_checking’, ‘pricing_calculating’, ‘finance_approving’, ‘completed’, ‘failed’, ‘awaiting_human’] def __init__(self, order_id): self.order_id order_id self.context {} self.machine Machine(modelself, statesOrderWorkflow.states, initial‘idle’) # 定义状态转移 self.machine.add_transition(trigger‘extract_order’, source‘idle’, dest‘order_extracting’) self.machine.add_transition(trigger‘inventory_ok’, source‘order_extracting’, dest‘pricing_calculating’) self.machine.add_transition(trigger‘inventory_fail’, source‘order_extracting’, dest‘awaiting_human’) self.machine.add_transition(trigger‘price_calculated’, source‘pricing_calculating’, dest‘finance_approving’) self.machine.add_transition(trigger‘finance_approved’, source‘finance_approving’, dest‘completed’) self.machine.add_transition(trigger‘finance_rejected’, source‘finance_approving’, dest‘awaiting_human’) def on_enter_inventory_checking(self): # 进入“核库存”状态时触发库存Agent的工作 event_bus.publish(AgentTaskEvent(task_type“check_inventory”, contextself.context, workflow_idself.order_id))协调者监听消息总线上的事件根据当前状态和事件类型触发相应的状态转移并发布新的任务事件。4.3 工具函数的实现与安全考量工具函数是连接LLM幻想世界和现实系统的桥梁必须健壮、安全。# 示例库存预留工具 def reserve_inventory(order_id: str, sku: str, quantity: int, warehouse_id: str) - dict: “”” 为指定订单预留库存。 参数: order_id: 销售订单号 sku: 产品代码 quantity: 预留数量必须为正整数 warehouse_id: 仓库代码 返回: {“success”: bool, “reservation_id”: str, “message”: str} “”” # 1. 输入验证 if quantity 0: return {“success”: False, “reservation_id”: None, “message”: “预留数量必须大于0”} # 2. 业务验证例如再次检查可用量避免并发冲突 available check_available_quantity(sku, warehouse_id) if available quantity: return {“success”: False, “reservation_id”: None, “message”: f“库存不足。可用{available}, 需求{quantity}”} # 3. 调用真正的ERP API或数据库操作在事务中 try: reservation_id erp_api.create_reservation(order_id, sku, quantity, warehouse_id) # 4. 记录审计日志 audit_log(action“reserve_inventory”, agent“InventoryAgent”, details{…}) return {“success”: True, “reservation_id”: reservation_id, “message”: “预留成功”} except ERPException as e: # 5. 异常处理与友好错误返回 logger.error(f“预留库存失败: {e}”) return {“success”: False, “reservation_id”: None, “message”: f“系统操作失败: {str(e)}”}安全与权限每个工具函数都应内置权限检查。例如approve_payment工具可能只允许“财务Agent”调用这需要在工具执行层或消息路由层进行身份校验。5. 落地实践中的挑战与解决方案理论设计很美好但真正落地时会遇到一系列棘手的问题。5.1 挑战一LLM输出的不稳定与解析失败问题LLM可能不按照你指定的JSON格式输出或者在工具调用参数中产生无效值。解决方案强化输出解析使用Pydantic模型来定义期望的输出结构并让LLM以JSON模式如OpenAI的response_format输出。解析失败时将错误信息反馈给LLM要求其重试通常设置最多2-3次重试。后置校验与清洗在工具被调用前对LLM生成的参数进行程序化校验。例如如果数量参数不是数字则用默认值或抛出明确错误给Agent上下文让它“重新思考”。提示词工程优化在系统提示词中强调“你必须输出纯JSON不要有任何额外解释”并在少样本示例中反复强化这一点。5.2 挑战二长流程中的上下文丢失与幻觉问题一个订单处理流程可能涉及十几次LLM调用和工具调用到了流程后期LLM可能会忘记早期的关键信息如客户特殊折扣甚至产生幻觉编造一个不存在的库存预留号。解决方案严格的上下文管理如前所述维护一个不断增长的、结构化的共享上下文对象。每次Agent被激活时只将它需要知道的部分上下文和全局关键结果传递给它而不是整个冗长的历史对话。这减少了无关信息的干扰。关键事实锚定对于流程中产生的关键结果如生成的订单号、预留ID、计算出的总价将其作为“已确认事实”存储在上下文里并在后续步骤的提示词中明确指出“根据上一步已确认的结果订单总价为5888.00元。请基于此进行财务审核。” 这样能有效对抗幻觉。审计轨迹引用允许Agent在思考时参考审计轨迹。例如提示词中可以写“关于库存预留请参考审计轨迹中InventoryAgent在10:05的操作结果。”5.3 挑战三错误处理与系统韧性问题某个工具调用因网络或下游系统故障失败或者LLM陷入了逻辑循环整个流程就会卡死。解决方案分层错误处理工具级工具函数返回标准化结构{success, data, error_message}并包含可重试的标识。Agent级Agent运行循环中捕获异常和工具失败根据策略决定是重试、请求澄清还是上报失败。工作流级协调者监控每个任务的状态和超时。对于失败任务根据预定义策略如“库存检查失败则转人工”驱动流程转向。看门狗与心跳为长时间运行的工作流实例设置看门狗计时器。如果某个状态停留时间过长则触发告警并尝试干预或重启该环节。人工接管接口设计良好的人机交互界面至关重要。当系统进入awaiting_human状态时能清晰地向操作员展示当前上下文、卡住的原因以及推荐的处置选项如“确认替代物料”、“特批价格”。5.4 挑战四性能与成本优化问题多轮LLM调用成本高长流程耗时可能达到分钟级。解决方案Agent的“短路”设计对于一些有明确规则、无需LLM复杂推理的步骤直接使用规则引擎或简单函数。例如“如果客户等级为VIP则自动批准信用额度小于10万的订单”这个判断完全可以用if-else实现无需调用财务Agent。上下文摘要与压缩在流程中后期将前期的详细对话历史总结成一段精简的摘要再传递给后续Agent大幅减少Token消耗。异步与并行化分析流程中的任务依赖关系。如果库存检查和生产能力评估互不依赖可以同时启动两个Agent并行执行最后由协调者汇总结果。模型分级使用对于创意生成、复杂决策等核心环节使用高性能大模型如GPT-4对于信息提取、简单分类等任务尝试使用更经济的小模型或专用模型。6. 效果评估与迭代优化系统上线后如何评估其好坏并持续改进6.1 建立多维度的评估体系不能只看“流程是否走通”需要更细致的指标流程成功率从开始到最终成功创建订单的百分比。人工干预率有多少流程需要人工介入介入点在哪个环节这是衡量自动化程度的关键。平均处理时间对比传统人工操作或旧系统时间缩短了多少单任务准确率针对每个Agent抽样评估其输出结果的准确性如销售Agent的信息抽取准确率财务Agent的定价计算正确率。成本平均处理一个订单所消耗的Token成本。6.2 构建“黄金数据集”与回归测试收集一批覆盖各种业务场景正常订单、特殊折扣、缺料、信用超标等的典型用户输入和期望的最终输出形成“黄金数据集”。每次对Agent提示词、工具或协作逻辑进行修改后都用这个数据集跑一遍回归测试确保修改没有破坏原有功能并且在新场景下有所提升。6.3 持续的提示词与流程调优多Agent系统是一个“活”的系统需要持续运营。日志分析定期分析审计日志找到高频失败点或人工干预点。例如发现财务审核环节经常因为某个特定促销规则而卡住那么就需要优化财务Agent的提示词或者为该规则创建一个专用工具。A/B测试对于关键Agent的提示词可以准备两个版本A/B在小流量上对比它们的成功率和准确性择优选用。业务规则同步当公司的定价政策、库存策略发生变化时不仅要更新后端系统还必须同步更新相关Agent的提示词、少样本示例和工具函数背后的逻辑。这需要建立跨部门业务、IT、AI团队的协同流程。从ERP系统出发设计LLM多Agent系统是一个将传统企业软件智能化、拟人化的深刻过程。它要求我们不仅懂技术更要懂业务。最大的体会是成功的关键不在于追求最前沿的模型而在于对业务流程的极致拆解、对Agent角色的精准定义以及设计出一套稳健、可观测、可干预的协作机制。这套系统上线后它处理的不仅仅是订单更是公司运行逻辑的数字镜像。我们团队在经历了最初几个月的“鸡飞狗跳”后现在这套系统已经稳定处理了公司超过30%的标准订单流程将业务人员从重复劳动中解放出来去处理那些真正需要人类智慧和经验的异常情况。这或许就是人机协同在未来企业中的常态。