从复杂Agent图到单一LLM：223节点工作流简化实战

在构建复杂业务自动化流程时你是否也曾被臃肿的 Agent 编排图所困扰数百个节点相互连接维护成本高逻辑追踪困难每次需求变更都如履薄冰。本文将分享一个真实的架构演进案例如何将一个包含 223 个节点的复杂 Agent 工作流成功替换为一个单一的开源大语言模型LLM。我们将深入探讨其背后的技术动机、实现路径、核心代码以及迁移过程中的关键考量为面临类似架构复杂性的开发者提供一套可复用的简化方案。1. 背景与核心概念从复杂图到统一智能体在深入技术细节之前我们需要理解几个核心概念以及它们在此次架构演进中的角色。1.1 什么是 Agent Graph智能体图Agent Graph或称工作流图是一种用于编排多个 AI 智能体Agent协同完成复杂任务的架构模式。每个节点代表一个具有特定功能的智能体如数据提取、分析、决策、执行节点间的连线定义了数据流和控制流。例如一个电商客服自动化流程可能包含“意图识别”、“查询数据库”、“生成回复”、“情感分析”等多个串联或并联的 Agent 节点。传统 Agent Graph 的典型问题维护成本高节点越多依赖关系越复杂调试和更新越困难。逻辑碎片化业务逻辑被分散到数十甚至上百个节点中难以形成全局视图。资源开销大每个节点可能是一个独立的微服务或函数带来额外的网络延迟、计算资源消耗和部署复杂度。灵活性差增加新功能或调整流程往往需要修改图结构牵一发而动全身。1.2 什么是 LLM大语言模型LLMLarge Language Model是一种基于海量文本数据训练出的深度学习模型如 GPT、LLaMA、ChatGLM 等。它具备强大的语言理解、生成、推理和代码能力。一个足够强大的 LLM 可以被视为一个“通用任务处理器”通过精心设计的提示词Prompt它可以替代许多单一功能的 Agent。LLM 作为统一智能体的潜力功能集成一个模型可以同时处理分类、总结、翻译、代码生成等多种任务。上下文理解拥有长上下文窗口的 LLM 能够记住整个会话历史和多轮交互做出更连贯的决策。指令跟随通过系统提示词System Prompt和思维链Chain-of-Thought等技术可以精确引导 LLM 执行复杂、多步骤的流程。1.3 演进的核心思想从“编排”到“内化”本次架构演进的核心是将原本通过外部图结构“编排”的分散逻辑“内化”到一个统一的 LLM 中。这并非简单的功能替换而是思维范式的转变从设计一个由许多“专家”组成的团队每个专家只做一件事转变为培养一个“通才”通过清晰的指令完成一系列任务。2. 环境准备与版本说明在开始重构之前需要搭建合适的环境。本文以 Python 为主要语言使用流行的开源 LLM 框架进行演示。基础环境操作系统Ubuntu 20.04 LTS 或 macOS (Apple Silicon) / Windows 11 with WSL2。本文命令以 Linux/macOS 为例。Python版本 3.9 或 3.10。推荐使用conda或venv创建虚拟环境。包管理工具pip。核心依赖库我们将使用litellm作为统一的 LLM 调用代理它支持众多开源和闭源模型。同时使用pydantic进行结构化输出。# 创建并激活虚拟环境 python -m venv llm_agent_env source llm_agent_env/bin/activate # Windows: llm_agent_env\Scripts\activate # 安装核心依赖 pip install litellm pydanticLLM 模型选择我们选择Mixtral 8x7B Instruct或成本更低的Llama 3 8B Instruct作为示例开源模型。你可以通过本地部署Ollama, vLLM或云服务Together AI, Groq来调用。本文假设你已有一个可访问的 LLM API 端点。# 例如如果你使用 Ollama 在本地运行 Llama 3 ollama pull llama3:8b-instruct-q4_K_M # 运行模型服务默认端口 11434 ollama serve3. 核心原理与设计模式拆解将图结构内化为 LLM 提示词需要精心的设计。以下是几种关键模式。3.1 系统提示词System Prompt作为“总控制器”系统提示词定义了 LLM 的角色、能力和行为规范相当于整个工作流的“宪法”或“总控制器”。它需要清晰地描述任务范围、输出格式、处理逻辑和禁忌。旧图逻辑伪代码:节点1接收用户输入 - 节点2分类意图 - 节点3根据意图路由- [节点4A处理查询 节点4B处理投诉...]新 LLM 提示词设计system_prompt 你是一个全能型业务助手。请严格按照以下步骤处理用户请求 1. **意图识别**首先判断用户意图类别包括产品查询、技术支持、订单操作、投诉建议。 2. **信息提取**从用户输入中提取关键实体如订单号、产品名称、问题描述。 3. **逻辑执行** - 如果是“产品查询”请模拟查询数据库并返回产品详情价格、库存、规格。 - 如果是“技术支持”请根据问题描述提供分步骤的排查建议。 - 如果是“订单操作”请确认操作类型查询、取消、修改并模拟执行。 - 如果是“投诉建议”请总结用户诉求并生成一封标准格式的安抚邮件草稿。 4. **输出格式**最终输出必须是一个有效的 JSON 对象包含字段intent, entities, response, next_step。 请一步一步思考。 这个单一的提示词内化了原先图中多个分类和路由节点的功能。3.2 结构化输出Structured Output与函数调用Function CallingLLM 的自由文本输出难以被下游程序消费。通过强制结构化输出我们可以获得稳定、可解析的结果。使用 Pydantic 定义输出模式from pydantic import BaseModel, Field from typing import List, Optional class CustomerServiceResponse(BaseModel): intent: str Field(description识别的用户意图) entities: dict Field(description提取的关键信息字典) response: str Field(description给用户的直接回复内容) next_step: str Field(description建议的系统下一步操作) confidence: float Field(description本次处理的置信度, ge0, le1) # 在调用 LLM 时通过 litellm 的 response_format 参数或特定模型的工具调用功能来约束输出。对于需要执行具体代码操作如真正查询数据库的场景可以利用 LLM 的函数调用Function Calling能力。我们将外部工具函数的定义暴露给 LLM由 LLM 决定在何时、以何种参数调用哪个函数。import litellm from litellm import completion # 1. 定义可供 LLM 调用的工具函数 def query_product_db(product_name: str) - dict: 模拟查询产品数据库 # 这里应该是真实的数据库查询逻辑 mock_db {手机: {price: 2999, stock: 100}} return mock_db.get(product_name, {error: product not found}) def create_service_ticket(issue: str, customer_id: str) - str: 模拟创建工单系统 return fTicket-CREATED-{customer_id} # 2. 将函数描述告知 LLM tools [ { type: function, function: { name: query_product_db, description: 根据产品名称查询产品详情价格、库存, parameters: { type: object, properties: { product_name: {type: string, description: 产品名称} }, required: [product_name] } } }, { type: function, function: { name: create_service_ticket, description: 为用户问题创建技术支持工单, parameters: { type: object, properties: { issue: {type: string, description: 问题描述}, customer_id: {type: string, description: 客户ID} }, required: [issue, customer_id] } } } ] # 3. 调用 LLM并允许其触发工具调用 response completion( modelollama/llama3:8b-instruct-q4_K_M, # 或你的模型端点 messages[{role: user, content: 我想了解一下你们最新款手机的价格和库存情况。}], toolstools, tool_choiceauto, # 让模型决定是否调用工具 )LLM 的回复可能会包含一个要求调用query_product_db的请求你的程序接收到这个请求后执行真实函数再将结果返回给 LLM 进行总结最终生成给用户的回复。这个过程将原先图中“LLM节点”-“API调用节点”的链路内化为 LLM 自主决策的循环。3.3 思维链Chain-of-Thought与少样本示例Few-Shot对于复杂推理要求 LLM “一步一步思考”Chain-of-Thought可以显著提高准确性。在系统提示词中加入请一步一步思考。或Let‘s think step by step.即可。对于格式复杂或容易出错的场景在提示词中提供少样本示例Few-Shot Examples是极其有效的方法。few_shot_prompt 请根据用户输入提取信息并生成JSON。 示例1 用户 “我的订单#ORD-12345怎么还没发货” 输出 {intent: 订单查询, order_id: ORD-12345, action: check_status} 示例2 用户 “手机屏幕碎了怎么保修” 输出 {intent: 技术支持, device: 手机, issue: 屏幕损坏, action: create_repair_ticket} 现在请处理新的用户输入 用户 “{user_input}” 输出 通过提供几个例子LLM 能快速掌握我们期望的输出格式和内容映射规则。4. 完整实战案例重构一个客户服务流程假设我们有一个旧的客户服务 Agent Graph包含以下节点输入解析-情感分析-意图分类-信息提取-路由-知识库查询/工单创建/订单查询-回复生成-输出格式化。我们将它重构为一个单一 LLM 流程。4.1 定义统一的数据模型与工具首先我们定义整个流程中流转的核心数据结构和工具函数。# data_models.py from pydantic import BaseModel, Field from typing import Optional, List from enum import Enum class IntentEnum(str, Enum): PRODUCT_QUERY 产品查询 TECH_SUPPORT 技术支持 ORDER_OPERATION 订单操作 COMPLAINT 投诉建议 class ExtractedEntities(BaseModel): product_name: Optional[str] None order_id: Optional[str] None customer_id: Optional[str] None issue_description: Optional[str] None class AgentResponse(BaseModel): LLM 最终输出的结构化格式 intent: IntentEnum entities: ExtractedEntities immediate_response: str internal_actions: List[str] Field(description需要系统执行的动作列表如 call_query_product_db) sentiment: str Field(description用户情绪positive, neutral, negative) # tools.py import json from datetime import datetime def query_product_db(product_name: str) - dict: 模拟数据库查询 print(f[TOOL CALL] query_product_db with: {product_name}) # 模拟数据库 products { 手机: {price: 2999, stock: 50, spec: 6.7英寸 OLED}, 笔记本电脑: {price: 8999, stock: 20, spec: 16GB RAM, 1TB SSD}, } return products.get(product_name, {error: Product not found}) def create_support_ticket(customer_id: str, issue: str) - str: 模拟创建工单 print(f[TOOL CALL] create_support_ticket for {customer_id}: {issue}) ticket_id fTICKET-{datetime.now().strftime(%Y%m%d%H%M%S)} # 这里应该调用真实的工单系统API return ticket_id def check_order_status(order_id: str) - dict: 模拟订单状态查询 print(f[TOOL CALL] check_order_status: {order_id}) # 模拟订单系统 return {order_id: order_id, status: shipped, estimated_delivery: 2023-10-30}4.2 构建核心 LLM 智能体引擎接下来创建核心的智能体类它整合了提示词工程、工具调用和响应解析。# unified_agent.py import litellm from litellm import completion from data_models import AgentResponse, IntentEnum from tools import query_product_db, create_support_ticket, check_order_status import json import re class UnifiedCustomerServiceAgent: def __init__(self, model: str ollama/llama3:8b-instruct-q4_K_M): self.model model self.system_prompt self._build_system_prompt() self.tools self._define_tools() def _build_system_prompt(self) - str: return 你是一个高级全栈客户服务AI。请按以下步骤处理对话 1. **理解与感知**分析用户输入的情感和核心诉求。 2. **识别与提取**判断意图产品查询、技术支持、订单操作、投诉建议并提取关键实体产品名、订单号等。 3. **规划与执行**思考需要调用哪些工具函数来获取信息或执行操作。如果需要请请求调用工具。 4. **综合与回复**基于所有信息包括工具返回结果生成对用户友好、准确、有帮助的回复。 5. **结构化输出**你的最终输出必须是以下JSON格式且仅包含这个JSON对象 { intent: 识别出的意图, entities: {product_name: ..., order_id: ..., ...}, immediate_response: 给用户的文本回复, internal_actions: [action1, action2], sentiment: positive/neutral/negative } 请一步一步思考。如果用户输入模糊请通过提问来澄清。 def _define_tools(self) - list: return [ { type: function, function: { name: query_product_db, description: 查询产品详细信息包括价格、库存和规格。, parameters: { type: object, properties: { product_name: {type: string} }, required: [product_name] } } }, { type: function, function: { name: create_support_ticket, description: 为用户报告的问题创建技术支持工单。, parameters: { type: object, properties: { customer_id: {type: string}, issue: {type: string} }, required: [customer_id, issue] } } }, { type: function, function: { name: check_order_status, description: 根据订单号查询订单的最新状态和预计送达时间。, parameters: { type: object, properties: { order_id: {type: string} }, required: [order_id] } } } ] def _execute_tool(self, tool_name: str, tool_args: dict) - str: 执行被LLM调用的工具 if tool_name query_product_db: result query_product_db(**tool_args) elif tool_name create_support_ticket: result create_support_ticket(**tool_args) elif tool_name check_order_status: result check_order_status(**tool_args) else: result {error: fUnknown tool: {tool_name}} return json.dumps(result, ensure_asciiFalse) def process(self, user_input: str, conversation_history: list None) - AgentResponse: 处理单轮用户输入 messages [{role: system, content: self.system_prompt}] if conversation_history: messages.extend(conversation_history) messages.append({role: user, content: user_input}) # 第一轮调用让LLM思考并可能请求调用工具 response completion( modelself.model, messagesmessages, toolsself.tools, tool_choiceauto, ) message response.choices[0].message final_content message.content # 处理工具调用请求 if hasattr(message, tool_calls) and message.tool_calls: print(f检测到工具调用请求: {message.tool_calls}) for tool_call in message.tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) tool_result self._execute_tool(tool_name, tool_args) # 将工具执行结果作为新的上下文追加 messages.append(message) # 追加包含工具调用的助理消息 messages.append({ role: tool, content: tool_result, tool_call_id: tool_call.id }) # 第二轮调用让LLM基于工具结果生成最终回复 second_response completion( modelself.model, messagesmessages, ) final_content second_response.choices[0].message.content # 从最终回复中解析JSON json_str self._extract_json(final_content) try: parsed_response AgentResponse.model_validate_json(json_str) return parsed_response except Exception as e: print(fJSON解析失败: {e}, 原始内容: {final_content}) # 优雅降级返回一个包含错误信息的默认响应 return AgentResponse( intentIntentEnum.TECH_SUPPORT, entities{}, immediate_response系统处理您的请求时遇到一点小问题请稍后再试或联系人工客服。, internal_actions[error_parsing_response], sentimentneutral ) def _extract_json(self, text: str) - str: 从可能包含额外文本的回复中提取JSON字符串 match re.search(r\{.*\}, text, re.DOTALL) return match.group(0) if match else {}4.3 运行与验证创建一个主程序来测试我们的统一智能体。# main.py from unified_agent import UnifiedCustomerServiceAgent def main(): agent UnifiedCustomerServiceAgent() test_cases [ 你们那款卖2999的手机还有货吗, 我的订单#ORD-20231025-001什么时候能送到, 刚买的电脑开不了机电源灯都不亮怎么办, 我要投诉快递员态度极其恶劣 ] for query in test_cases: print(f\n{*50}) print(f用户输入: {query}) print(f{-*50}) response agent.process(query) print(f识别意图: {response.intent}) print(f提取实体: {response.entities}) print(f情绪分析: {response.sentiment}) print(f内部动作: {response.internal_actions}) print(f生成回复: {response.immediate_response}) print(f{*50}) if __name__ __main__: main()预期输出示例 用户输入: 你们那款卖2999的手机还有货吗 -------------------------------------------------- [TOOL CALL] query_product_db with: 手机 识别意图: IntentEnum.PRODUCT_QUERY 提取实体: {product_name: 手机} 情绪分析: neutral 内部动作: [call_query_product_db] 生成回复: 您好您询问的这款手机价格2999元目前库存为50件。规格是6.7英寸OLED屏幕。请问您是否需要了解更多信息或协助下单 通过运行你可以看到一个简单的用户查询触发了工具调用并生成了包含数据库查询结果的、结构化的完整回复。原先需要多个节点协作的流程现在由一个 LLM 协调完成。5. 常见问题与排查思路在从 Agent Graph 迁移到单一 LLM 的过程中你可能会遇到以下典型问题。问题现象常见原因解决思路LLM 不遵循输出格式1. 系统提示词约束力不足。2. 模型能力有限。3. 少样本示例不够清晰。1. 在提示词中强调“必须”、“只能输出JSON”。2. 换用指令跟随能力更强的模型如 GPT-4, Claude 3, 或微调后的开源模型。3. 提供更精确的 Few-Shot 示例。使用response_format参数如果模型支持。工具调用不稳定或错误1. 工具描述不清晰。2. LLM 无法正确解析参数。3. 函数签名与描述不匹配。1. 优化工具函数的description和parameters描述使其无歧义。2. 在提示词中要求 LLM “逐步推理”先思考再调用。3. 在代码中加入参数验证和后处理逻辑。处理复杂逻辑时效果下降单一提示词过长或逻辑过于复杂超出模型上下文处理能力。1. 采用“分治”策略设计一个“主控”LLM 来分解任务调用不同的“子任务”LLM或函数。2. 使用 LangChain 或 LlamaIndex 等框架的“Router”模式。注意这会在一定程度上回归图的思想但节点数会大幅减少。响应速度变慢1. 模型本身推理速度慢。2. 需要多轮工具调用串行等待。1. 考虑使用推理速度更快的模型或服务如 Groq。2. 优化工具设计尽可能一次调用获取更多信息。3. 对于可并行的工具调用探索异步调用机制。难以处理严格的状态逻辑LLM 本质是无状态的复杂的状态机如严格的订单审批流程难以仅靠提示词保证。1. 将核心状态机逻辑保留在外部代码中LLM 作为“建议者”而非“执行者”。2. 使用更高级的框架如 Microsoft Autogen, CrewAI来管理多 Agent 对话状态但其复杂度低于维护 200 节点的图。6. 最佳实践与工程建议基于实战经验以下建议能帮助你更平稳、高效地完成此类架构迁移。6.1 迁移策略渐进式而非颠覆式不要试图一次性替换整个 223 节点的图。建议识别边界从图中识别出逻辑相对独立、功能明确的子图例如“订单查询”子图。逐个击破为这个子图设计对应的 LLM 提示词和工具集实现一个功能对等的“微智能体”。并行运行与对比让新旧两套逻辑并行运行一段时间对比输出结果、性能和稳定性。替换与下线确认新逻辑稳定可靠后将流量切换到新的 LLM 微智能体并下线对应的旧子图节点。迭代循环重复以上步骤逐步蚕食整个复杂图。6.2 提示词工程模块化与版本化模块化设计将系统提示词拆解为角色定义、任务步骤、输出格式、示例等模块便于维护和组合。版本控制像管理代码一样管理你的提示词。使用 Git 对提示词模板进行版本控制便于回滚和 A/B 测试。持续评估建立评估体系用一批标准测试用例定期评估 LLM 输出的质量监控提示词修改带来的影响。6.3 工程化与可观测性日志记录详细记录 LLM 的输入提示词、输出包括中间的工具调用请求和结果。这是调试和优化最重要的依据。设置超时与重试LLM API 调用可能失败或超时必须设置合理的超时时间并实现重试机制注意幂等性。限流与降级对 LLM 服务调用进行限流防止意外流量打垮服务。设计降级方案例如在 LLM 服务不可用时回退到基于规则的简单逻辑。成本监控如果使用按 token 计费的云服务必须密切监控调用成本和 token 消耗优化提示词长度。6.4 安全与合规输入输出过滤对用户输入和 LLM 输出进行严格的过滤和审查防止提示词注入、敏感信息泄露或生成有害内容。工具调用沙箱化对于执行数据库写入、发送邮件、调用外部 API 等有副作用的工具函数必须进行严格的权限校验和参数审查避免 LLM 被恶意引导执行危险操作。数据隐私确保发送给 LLM 服务的数据不包含用户个人身份信息PII或已进行脱敏处理。对于开源模型本地部署此风险较低。7. 总结将 223 个节点的 Agent Graph 替换为单一开源 LLM并非宣称“图模式已死”而是展示了在 LLM 能力边界内通过提示词工程、函数调用和结构化输出可以极大地简化系统架构、降低维护复杂度。这种模式特别适用于逻辑以语言理解和生成为核心、对绝对确定性要求相对宽松的业务场景。关键收获思维转变从“如何设计流程”转向“如何描述任务”让 LLM 承担内部的流程编排。技术聚焦掌握强大的系统提示词设计、少样本学习、思维链和函数调用技术是成功的关键。渐进迁移采用“分而治之”的渐进策略能有效控制风险。工程配套没有良好的日志、监控、评估和安全措施LLM 应用难以在生产环境稳定运行。下一步学习路线深入提示词工程学习更高级的技巧如 ReAct、Self-Consistency、Tree of Thoughts。探索智能体框架了解 LangChain、LlamaIndex、Semantic Kernel 等框架它们提供了更丰富的模式如 Agent Executor, Router来管理复杂程度超出单一提示词的任务。模型微调对于垂直领域考虑使用业务数据对开源模型进行微调Fine-tuning以获得更精准、更可控的表现。评估与优化建立自动化的评估流水线持续优化提示词和工具集。架构的简化带来了可维护性的巨大提升但也对提示词设计、异常处理和安全提出了新的挑战。希望本文提供的思路、代码和实践经验能帮助你在探索大模型赋能业务自动化的道路上找到复杂度与效能的最佳平衡点。