Agent系统设计七宗罪:从while循环到生产级自主决策体
2026/9/19 22:14:32 网站建设 项目流程

1. 这不是“加个while循环就叫Agent”——一个被严重低估的系统设计战场

你有没有在凌晨三点改完第十七版prompt,把LLM输出硬塞进JSON解析器,结果又因为多了一个空格、少了一个逗号、或者返回了一段“好的,我明白了”而崩溃?你是不是也试过用while循环反复调用模型,直到它“终于吐出正确格式”,然后在日志里看到连续23次重试、平均响应延迟飙到8.7秒、错误率31%?这不是你的问题——这是整个行业对“Agent”这个词最普遍、最危险的误读。标题里说的“别只会给LLM包一层while循环”,根本不是在批评初学者,而是在戳破一个广泛存在的认知泡沫:把LLM当做一个可无限重试的黑盒API来调用,本质上仍是单次推理思维,和真正的Agent设计毫无关系。真正意义上的Agent,是具备目标分解能力、工具调度意识、状态记忆机制、失败回溯逻辑、执行路径规划、容错降级策略和可观测反馈闭环的自主决策体。它不依赖“运气好一次就成”,而是构建一套鲁棒的执行骨架,让LLM在其中扮演“专家顾问”而非“全栈工人”。这篇文章要拆解的,正是这七个无法绕开的设计取舍——它们不是技术选型清单,而是你在画第一张架构图之前,必须亲手写下的七道判断题。每个选择背后,都直接决定你的Agent是能稳定跑通生产环境的业务组件,还是只能在demo视频里闪亮5秒的玩具。我带团队落地过6个跨行业Agent项目,从金融风控决策链到工业设备远程诊断助手,踩过的坑足够填满三本笔记本。下面这七个取舍,每一个都来自真实压测现场、客户投诉邮件和凌晨三点的紧急回滚操作记录。

2. 七个核心设计取舍:为什么选A而不是B,从来不是技术问题

2.1 取舍一:目标驱动 vs. 指令驱动——你的Agent到底在“做什么”,还是在“听什么”

绝大多数初学者写的Agent,本质是“指令驱动”的:用户输入一句话,Agent把它喂给LLM,LLM吐出一段文字,Agent再把文字包装成回复。整个流程像一个高级版的Echo服务。但真正的Agent必须是“目标驱动”的:它接收的是一个可验证的终态目标(例如:“查出客户张三过去三个月所有逾期未还款项,并生成催收优先级排序表”),而不是一句模糊指令(例如:“帮我查查张三的账”)。这个区别看似微小,实则决定了整个系统的底层架构。

目标驱动意味着Agent必须内置一套目标分解引擎。它需要把高层目标拆解为原子任务序列:先调用CRM API查客户ID,再用ID查账单流水,过滤出逾期项,计算逾期天数和金额权重,最后调用报表生成服务。每一步都需定义输入约束、输出契约、失败阈值和替代路径。而指令驱动下,LLM被迫承担全部分解工作——这不仅极大增加prompt复杂度,更导致不可控的幻觉风险。我们曾在一个保险核保Agent中发现,当用户说“看看这个保单能不能续保”,LLM有17%概率自行虚构一个不存在的“核保规则库”并据此编造结论,只因它没被明确告知“必须调用规则查询接口”。

提示:目标驱动的标志是——Agent内部存在一个显式的、可序列化的任务图(Task Graph),而非隐式的、依赖LLM自由发挥的文本流。这个图可以是静态预定义的(如金融场景的SOP流程),也可以是动态生成的(如客服场景的意图树),但绝不能缺失。

实操中,我们采用“目标-动作-约束”三元组建模法:

  • 目标{ "type": "generate_collection_plan", "customer_id": "CUST-2024-8871", "time_window": "P3M" }
  • 动作[ { "tool": "crm_search", "params": { "id": "CUST-2024-8871" } }, { "tool": "billing_api", "params": { "customer_id": "CUST-2024-8871", "start_date": "2024-04-01" } } ]
  • 约束{ "max_retries": 2, "timeout_ms": 12000, "fallback_tool": "manual_review_queue" }

这种结构让调试变得极其清晰:当任务失败时,你能准确定位是CRM接口超时、还是账单API返回了非预期字段,而不是在LLM的千字输出里大海捞针。更重要的是,它天然支持审计——每个动作都有明确的输入输出日志,满足金融、医疗等强监管行业的留痕要求。

2.2 取舍二:显式状态管理 vs. 隐式上下文拼接——你的Agent记得住自己做过什么吗

“让LLM记住上下文”是另一个高危误区。很多方案把历史对话轮次、工具调用结果、中间变量一股脑塞进system prompt或context window,指望LLM自己“理解”当前进展。问题在于:LLM没有真正的记忆,只有注意力权重;它无法区分“用户刚说的”和“三轮前调用的API返回值”,更无法在长上下文中精准定位关键状态变更点。我们在电商售后Agent压测中发现,当对话超过12轮,LLM对“已生成退货单号RD-2024-9912”的引用错误率高达43%,常把它错记为“待审核”而非“已发货”。

真正的状态管理必须是显式、结构化、可验证的。我们采用三层状态架构:

  • 会话层(Session State):存储用户身份、会话ID、初始目标、全局超时设置。使用Redis哈希结构,TTL设为24小时。
  • 执行层(Execution State):记录当前任务图进度、已完成动作、待执行动作、各动作的输入输出快照。采用JSON Schema严格校验,每次状态更新都触发schema验证。
  • 记忆层(Memory State):长期记忆(如用户偏好、历史决策模式)存入向量数据库,但仅用于增强检索,不参与核心流程控制。

关键设计点在于状态变更的原子性。每次工具调用后,Agent必须执行三个不可分割的操作:

  1. 解析工具返回结果,提取结构化字段(如{"status":"success","tracking_number":"SF123456789CN"}
  2. 更新执行层状态(将shipping_task标记为completed,存入tracking_number
  3. 生成下一轮决策提示(基于新状态,而非原始上下文)

这个过程完全脱离LLM——它只负责根据当前状态生成下一步动作描述,而状态更新由确定性代码完成。我们甚至在状态更新后加入校验钩子:若检测到tracking_number字段为空但状态标记为completed,立即触发告警并回滚。这种设计让Agent具备了传统软件系统的可靠性基因,而不是AI模型的随机性。

2.3 取舍三:工具编排 vs. 工具调用——你的Agent是调度员,还是搬运工

把一堆API封装成函数,然后让LLM“选一个调用”,这只是工具调用(Tool Calling);而工具编排(Tool Orchestration)是让Agent理解工具间的依赖关系、数据流向、错误传播路径和资源约束。前者是LLM的附加能力,后者是Agent的核心能力。

举个典型反例:一个招聘筛选Agent需要“解析简历PDF→提取技能关键词→匹配JD要求→生成评分报告”。如果只做工具调用,LLM可能先调用PDF解析,再调用JD匹配,最后才想起要提取技能——但JD匹配工具根本无法处理原始PDF,必须等待技能列表。这种顺序错误在真实场景中频繁发生,导致大量无效调用和超时。

我们的编排方案强制引入数据契约(Data Contract)

  • 每个工具声明其input_schemaoutput_schema
  • Agent运行时构建数据流图(Data Flow Graph),自动检测依赖环
  • 执行前进行静态分析:若工具A输出skills: string[],而工具B输入要求skills: string[],则建立A→B边

更关键的是错误传播控制。当PDF解析失败时,传统方案会让LLM重新尝试或放弃;而编排系统会检查:该失败是否影响后续所有节点?如果是,则触发降级路径(如启用OCR备用通道);如果仅影响技能匹配,则隔离该分支,用通用模板生成报告。我们在物流追踪Agent中实现过三级降级:

  • L1:主快递API失败 → 切换至聚合查询平台
  • L2:聚合平台无数据 → 启动用户位置反查(基于发货地址估算)
  • L3:所有外部源失效 → 返回预置的“运输中”状态+预计时效区间

这种编排能力无法靠LLM推理获得,必须由确定性代码实现。它让Agent从“尽力而为”变成“可控交付”。

2.4 取舍四:确定性决策 vs. 概率性生成——你的Agent敢为自己的选择负责吗

LLM输出本质上是概率分布采样,这意味着同一输入可能产生不同输出。但在生产环境中,你无法接受“今天生成催收话术A,明天生成话术B,后天生成法律风险提示”。Agent必须在LLM的不确定性之上,构建确定性决策层

我们的方案是决策-执行分离架构

  • 决策层(Deterministic):基于规则引擎(Drools)或决策树(scikit-learn导出的JSON)生成动作指令(Action Command),格式严格为{"tool":"send_sms","params":{"template_id":"COLL-001","recipient":"+86138****1234"}}
  • 执行层(Probabilistic):LLM仅负责将动作指令渲染为用户可见内容(User-Facing Content),如短信文案:“张三先生您好,您尾号1234的信用卡已逾期15天,请及时还款。【XX银行】”

这个分离带来三个关键收益:

  1. 可审计性:所有决策日志都是结构化JSON,可直接入库分析,无需NLP解析
  2. 可测试性:决策层可用单元测试覆盖100%分支,执行层只需测试渲染模板
  3. 可干预性:当发现某类催收决策效果差,可直接修改规则引擎,无需重训LLM

我们曾用此架构将金融催收Agent的合规审查通过率从72%提升至99.8%——因为所有决策都源于预审通过的规则集,LLM只负责“润色”,不参与“定性”。

注意:不要试图用temperature=0消除LLM随机性。实测显示,即使temperature=0,在长文本生成中仍存在token级波动。真正的确定性必须来自架构分层,而非参数调节。

2.5 取舍五:同步执行 vs. 异步编排——你的Agent能处理“等一等”吗

现实世界充满异步操作:发邮件要等SMTP响应、调用第三方API可能超时、用户需要时间思考回复。但多数LLM Agent设计默认采用同步阻塞模型,导致整个流程卡死。更糟的是,有些方案用while循环“轮询”异步任务状态,造成大量无效请求和资源浪费。

我们采用事件驱动异步编排

  • 每个异步任务启动时,生成唯一task_id,存入Redis并设置TTL
  • Agent立即返回{"status":"pending","task_id":"TASK-2024-7788"}
  • 独立Worker监听任务完成事件(如Kafka消息),更新Redis状态
  • 用户下次请求携带task_id,Agent查询状态并决定继续或返回结果

这套机制的关键创新在于状态机嵌套。一个顶层任务(如“完成整套开户流程”)包含多个子任务(身份证OCR、人脸识别、征信查询),每个子任务有自己的状态机。当人脸识别超时时,系统不会中断整个开户,而是标记该子任务为timeout,并启动人工复核流程,其他子任务继续运行。

在证券开户Agent中,这套机制将平均开户时长从23分钟降至8.4分钟,失败率下降67%——因为不再需要用户全程等待,也不再因单点故障导致全链路失败。

2.6 取舍六:轻量路由 vs. 重量推理——你的Agent需要多聪明才能做选择

很多Agent框架把“工具选择”交给LLM完成,认为“大模型懂一切”。但实际中,90%的路由决策是高度结构化的:根据用户意图类型(intent)、当前状态(state)、可用工具集合(tools),就能确定下一步。让LLM处理这种确定性逻辑,既浪费算力,又引入不必要风险。

我们的双层路由机制

  • L1 轻量路由(<5ms):基于意图识别模型(小型BERT微调)+ 状态机规则,快速匹配95%常见路径。例如,当intent=check_balance且state=logged_in,直接路由至banking_api.get_balance
  • L2 重量推理(~2s):仅当L1无法匹配(如遇到新意图、状态异常、工具冲突)时,才将上下文摘要送入LLM进行深度推理。

这个设计让87%的请求绕过LLM,显著降低延迟和成本。更重要的是,它实现了路由可解释性:运维人员能清晰看到“为什么选这个工具”,而不是面对LLM的黑盒输出抓耳挠腮。我们在政务咨询Agent中统计过,L1路由准确率达99.2%,而L2推理的准确率仅83.7%——证明简单规则往往比复杂推理更可靠。

实操心得:L1路由的规则库必须支持热更新。我们用Lua脚本编写规则,部署在Redis中,修改后5秒内生效,无需重启服务。这比训练新模型快三个数量级。

2.7 取舍七:可观测闭环 vs. 黑盒执行——你的Agent出了问题,你能找到根因吗

最后也是最致命的取舍:是否构建完整的可观测闭环。没有日志、没有指标、没有链路追踪的Agent,就像没有仪表盘的飞机——飞得再高,一旦出事就是灾难。

我们的可观测体系包含四个维度:

  • Trace(链路追踪):使用OpenTelemetry,记录每个任务从接收目标到返回结果的完整路径,标注LLM调用、工具执行、状态更新等关键节点
  • Metric(指标监控):实时采集成功率、平均延迟、LLM token消耗、工具调用频次等27项核心指标
  • Log(结构化日志):所有日志为JSON格式,包含trace_idsession_idtask_idstep_nameduration_mserror_code等字段
  • Profile(性能剖析):对LLM调用进行token级分析,识别prompt膨胀、冗余上下文、低效few-shot等性能瓶颈

最关键的创新是根因自动关联。当某个任务失败时,系统自动关联:

  • 该任务的完整trace
  • 同一session的前序操作日志
  • 相同tool_id的最近10次调用指标趋势
  • 当前LLM实例的GPU显存占用率

在一次支付失败排查中,这套系统3分钟内定位到问题:不是LLM出错,而是支付网关在特定时段返回了非标准HTTP状态码(499),而我们的错误处理器未覆盖该码——这完全与LLM无关,但传统方案会误判为“模型不稳定”。

3. 实操落地:从取舍到代码——一个可运行的Agent骨架

3.1 架构全景图:七个取舍如何落地为代码模块

我们以一个简化的“智能会议纪要Agent”为例,展示如何将前述七个取舍转化为可运行代码。该Agent需完成:接收会议录音→转文字→识别关键决策项→生成待办事项→邮件发送给参会者。

├── core/ # 核心引擎(体现取舍1/2/4/6) │ ├── planner/ # 目标分解器(取舍1:目标驱动) │ │ └── task_graph.py # 生成任务图:transcribe → extract_decisions → generate_actions → send_email │ ├── state/ # 状态管理器(取舍2:显式状态) │ │ ├── session.py # 会话状态(Redis-backed) │ │ └── execution.py # 执行状态(JSON Schema validated) │ ├── router/ # 双层路由(取舍6:轻量vs重量) │ │ ├── intent_classifier.py # L1意图识别(小型BERT) │ │ └── llm_router.py # L2深度推理(仅fallback时调用) │ └── decision/ # 决策层(取舍4:确定性决策) │ └── rule_engine.py # Drools规则引擎配置 ├── tools/ # 工具编排(取舍3:工具编排) │ ├── speech_to_text.py # 转录工具(声明input/output schema) │ ├── nlp_extractor.py # 决策项提取(依赖transcribe输出) │ └── email_sender.py # 邮件发送(含降级:SMTP失败→企业微信) ├── exec/ # 执行引擎(取舍5:异步编排) │ ├── async_worker.py # Kafka消费者,处理异步任务完成事件 │ └── task_manager.py # 任务状态机,支持pending/running/completed/failed └── observability/ # 可观测性(取舍7:可观测闭环) ├── tracer.py # OpenTelemetry trace注入 ├── metrics.py # Prometheus指标暴露 └── logger.py # 结构化JSON日志

这个目录结构本身就是七个取舍的物理映射。注意core/目录下没有LLM调用代码——LLM只作为tools/中的一个可插拔组件存在,其调用由decision/层触发,输出由exec/层消费。这种解耦让每个模块可独立演进、测试和替换。

3.2 关键代码片段:状态管理与任务图生成

以下是planner/task_graph.py的核心逻辑,体现取舍1(目标驱动)和取舍2(显式状态):

from typing import List, Dict, Any import jsonschema from jsonschema import validate # 任务图Schema定义(取舍2:状态可验证) TASK_GRAPH_SCHEMA = { "type": "object", "properties": { "target": {"type": "string"}, "steps": { "type": "array", "items": { "type": "object", "properties": { "id": {"type": "string"}, "tool": {"type": "string"}, "depends_on": {"type": "array", "items": {"type": "string"}}, "input_schema": {"type": "object"}, "output_schema": {"type": "object"}, "max_retries": {"type": "integer", "minimum": 0}, "timeout_ms": {"type": "integer", "minimum": 100} } } } } } def generate_task_graph(goal: Dict[str, Any]) -> Dict[str, Any]: """ 根据目标生成任务图(取舍1:目标驱动) goal示例: {"type": "generate_meeting_minutes", "recording_url": "s3://bucket/meeting.mp3"} """ if goal["type"] == "generate_meeting_minutes": # 静态任务图(SOP流程) graph = { "target": "generate_meeting_minutes", "steps": [ { "id": "transcribe", "tool": "speech_to_text", "depends_on": [], "input_schema": {"recording_url": "string"}, "output_schema": {"text": "string", "duration_sec": "number"}, "max_retries": 2, "timeout_ms": 120000 }, { "id": "extract_decisions", "tool": "nlp_extractor", "depends_on": ["transcribe"], "input_schema": {"meeting_text": "string"}, "output_schema": {"decisions": "array", "action_items": "array"}, "max_retries": 1, "timeout_ms": 30000 }, { "id": "send_email", "tool": "email_sender", "depends_on": ["extract_decisions"], "input_schema": {"to_emails": "array", "content": "string"}, "output_schema": {"message_id": "string"}, "max_retries": 3, "timeout_ms": 60000 } ] } else: raise ValueError(f"Unsupported goal type: {goal['type']}") # 强制Schema验证(取舍2:状态可验证) try: validate(instance=graph, schema=TASK_GRAPH_SCHEMA) except jsonschema.ValidationError as e: raise RuntimeError(f"Invalid task graph: {e.message}") return graph

这段代码的关键在于:

  • 目标驱动generate_task_graph函数只接收结构化goal,不处理原始文本
  • 显式依赖depends_on字段明确定义执行顺序,避免LLM自由发挥
  • 契约先行:每个step声明input_schema/output_schema,为工具编排提供基础
  • 可验证性:JSON Schema校验确保任务图永远符合预期结构

3.3 状态更新与错误处理:让Agent真正“记得住”和“扛得住”

state/execution.py中的状态更新逻辑,体现取舍2(显式状态)和取舍3(工具编排):

import redis import json from datetime import datetime class ExecutionState: def __init__(self, session_id: str): self.redis = redis.Redis() self.session_id = session_id def update_step(self, step_id: str, status: str, input_data: dict = None, output_data: dict = None, error: str = None) -> bool: """ 原子性更新步骤状态(取舍2:状态变更原子性) """ # 1. 获取当前状态 key = f"state:{self.session_id}" current_state = self.redis.hgetall(key) # 2. 构建新状态(只更新指定step) step_key = f"step:{step_id}" new_step_data = { "status": status, "updated_at": datetime.now().isoformat(), "input": json.dumps(input_data) if input_data else None, "output": json.dumps(output_data) if output_data else None, "error": error } # 3. 原子性写入(Redis pipeline保证) pipe = self.redis.pipeline() pipe.hset(key, step_key, json.dumps(new_step_data)) pipe.expire(key, 86400) # TTL 24h pipe.execute() # 4. 触发状态变更钩子(取舍3:错误传播控制) if status == "failed" and error: self._handle_step_failure(step_id, error) return True def _handle_step_failure(self, step_id: str, error: str): """ 错误传播控制(取舍3:工具编排的错误处理) """ # 查询该step的依赖关系 task_graph = self._get_task_graph() failed_step = next((s for s in task_graph["steps"] if s["id"] == step_id), None) if not failed_step: return # 根据错误类型和依赖关系决定降级策略 if step_id == "transcribe" and "timeout" in error: # L1降级:切换至备用ASR服务 self._trigger_fallback("transcribe_backup") elif step_id == "email_sender" and "smtp_error" in error: # L2降级:切换至企业微信 self._trigger_fallback("wechat_notifier") else: # L3降级:标记为manual_review self.update_step("manual_review", "pending", input_data={"failed_step": step_id, "error": error})

这个实现确保:

  • 状态更新不可分割:Redis pipeline保证hsetexpire原子执行
  • 错误即刻响应:失败时立即触发降级,而非等待LLM下一轮推理
  • 降级策略可配置:不同工具、不同错误类型对应不同fallback路径

3.4 可观测性集成:让问题无所遁形

observability/tracer.py中的链路追踪,体现取舍7(可观测闭环):

from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter # 初始化Tracer(生产环境连接Jaeger或Zipkin) provider = TracerProvider() processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://jaeger:4318/v1/traces")) provider.add_span_processor(processor) trace.set_tracer_provider(provider) def start_agent_trace(session_id: str, goal_type: str): """ 启动Agent全链路追踪(取舍7:可观测闭环) """ tracer = trace.get_tracer(__name__) with tracer.start_as_current_span( "agent_execution", attributes={ "session_id": session_id, "goal_type": goal_type, "env": "production" } ) as span: # 记录关键事件 span.add_event("goal_received", {"goal": str(goal_type)}) # 在每个核心步骤中创建子span with tracer.start_as_current_span("task_planning") as plan_span: task_graph = generate_task_graph({"type": goal_type}) plan_span.add_event("task_graph_generated", {"step_count": len(task_graph["steps"])}) with tracer.start_as_current_span("state_initialization") as state_span: state = ExecutionState(session_id) state_span.add_event("state_initialized") # ... 其他步骤 ... span.add_event("execution_completed") return span.context.trace_id # 在工具调用中注入trace context def call_tool_with_trace(tool_name: str, params: dict, trace_id: str): tracer = trace.get_tracer(__name__) with tracer.start_as_current_span( f"tool_{tool_name}", attributes={"tool_name": tool_name, "params": str(params)} ) as span: # 注入trace_id到HTTP header(如调用外部API) headers = {"X-Trace-ID": trace_id} # ... 实际调用逻辑 ... span.add_event("tool_called", {"status": "success"})

这套追踪体系让每个Agent执行都生成一条完整链路,包含:

  • 顶层agent_executionspan(标识会话和目标)
  • 子span如task_planningstate_initialization(标识各模块耗时)
  • 工具调用span(标识外部依赖性能)
  • 关键事件(如goal_receivedexecution_completed

当出现延迟问题时,运维人员可直接在Jaeger UI中查看火焰图,精准定位是task_planning耗时过长(需优化规则引擎),还是tool_speech_to_text响应慢(需联系ASR供应商)。

4. 常见问题与避坑指南:那些没人告诉你的血泪教训

4.1 “LLM返回JSON格式错误”——这不是LLM的问题,是你的契约没定好

网络热词中高频出现的“修复llm返回json的java库”,暴露了一个根本性误解:试图用后处理修复LLM的输出,而不是在源头约束其行为。我们统计过,83%的JSON解析失败源于LLM生成了非法字符(如中文引号、未转义换行符)或结构偏差(如多出逗号、少括号),而非语义错误。

正确解法:在LLM调用前,用结构化Prompt + JSON Schema约束

你是一个严格的JSON生成器。请严格按照以下JSON Schema输出,不得添加任何额外字段、注释或说明文字: { "type": "object", "properties": { "action_items": { "type": "array", "items": { "type": "object", "properties": { "task": {"type": "string"}, "owner": {"type": "string"}, "due_date": {"type": "string", "format": "date"} }, "required": ["task", "owner", "due_date"] } } }, "required": ["action_items"] }

同时,在代码层强制校验:

import jsonschema from jsonschema import validate def safe_json_parse(llm_output: str, schema: dict) -> dict: try: data = json.loads(llm_output) validate(instance=data, schema=schema) return data except json.JSONDecodeError as e: # 记录原始输出用于debug logger.error(f"JSON decode error: {e}, raw_output: {llm_output[:200]}") raise ValueError("Invalid JSON format") except jsonschema.ValidationError as e: logger.error(f"JSON schema validation error: {e.message}") raise ValueError("JSON does not match schema")

实操心得:永远不要信任LLM的JSON输出。我们曾因一个未捕获的json.JSONDecodeError导致整个批处理任务静默失败,损失了27小时的数据同步。现在所有LLM JSON输出都经过双重校验:先json.loads(),再jsonschema.validate(),任一失败都触发告警。

4.2 “Agent执行terminated due to error”——错误处理不是写个try-catch就够了

这个报错信息在Hermes Agent等框架中很常见,根源往往是错误处理粒度太粗。把整个Agent流程包在一个try-catch里,一旦出错就全链路终止,完全违背了Agent应有的韧性。

正确解法:按任务图节点粒度处理错误:

def execute_task_graph(graph: dict, state: ExecutionState): for step in graph["steps"]: try: # 执行step result = call_tool(step["tool"], step["input"]) # 更新状态 state.update_step(step["id"], "completed", output_data=result) except ToolTimeoutError as e: # 节点级超时处理 state.update_step(step["id"], "failed", error=f"timeout: {str(e)}") if step.get("fallback_tool"): # 启动fallback fallback_result = call_tool(step["fallback_tool"], step["input"]) state.update_step(f"{step['id']}_fallback", "completed", output_data=fallback_result) else: # 标记为需人工介入 state.update_step("manual_intervention", "pending", input_data={"step_id": step["id"], "error": str(e)}) except Exception as e: # 兜底错误处理 state.update_step(step["id"], "failed", error=f"unknown: {str(e)}")

这种设计让Agent具备“局部失败,全局存活”的能力。即使某个工具永久失效,系统仍能通过fallback或人工介入继续推进。

4.3 “Dify的SQL查询内容太多导致LLM返回不稳定”——不是LLM不行,是你的数据没剪枝

Dify等低代码平台常遇到此问题:当SQL查询返回数百行数据,LLM prompt迅速膨胀,导致token超限、响应变慢、输出质量下降。这不是LLM缺陷,而是数据管道设计失误。

正确解法:在LLM之前插入智能数据剪枝层

def smart_prune_sql_result(sql_result: List[dict], max_rows: int = 10) -> List[dict]: """ 智能剪枝:保留关键行 + 统计摘要(取舍4:确定性决策) """ if len(sql_result) <= max_rows: return sql_result # 1. 保留首尾各2行(看开头和结尾) pruned = sql_result[:2] + sql_result[-2:] # 2. 添加统计摘要(确定性生成,非LLM) summary = { "total_rows": len(sql_result), "columns": list(sql_result[0].keys()) if sql_result else [], "sample_values": {k: [r[k] for r in sql_result[:3]] for k in sql_result[0].keys() if sql_result} if sql_result else {} } # 3. 将摘要作为独立字段加入结果 pruned.append({"__summary__": summary}) return pruned # 使用示例 raw_data = execute_sql("SELECT * FROM orders WHERE status='pending'") pruned_data = smart_prune_sql_result(raw_data) # LLM prompt中只包含pruned_data,而非原始大数据集

这个剪枝层完全确定性,不依赖LLM,却解决了90%的大数据量问题。我们在电商Agent中应用后,LLM响应延迟从平均4.2秒降至0.8秒,token消耗减少76%。

4.4 “Prompt injection attack to tool selection”——安全不是加个filter,是重构信任边界

NDSS 2026论文提到的prompt注入攻击,本质是利用LLM对指令的盲目服从。传统方案用正则过滤敏感词,但攻击者总能找到绕过方式(如base64编码、Unicode混淆)。

正确解法:实施零信任工具调用协议

  1. 指令白名单:LLM输出必须匹配预定义的动作模板,否则拒绝执行

    # 预定义模板 TOOL_CALL_TEMPLATES = { "search_knowledge": r'{"tool": "search_knowledge", "params": {"query": "[^"]+"}}', "send_email": r'{"tool": "send_email", "params": {"to": "[^"]+", "subject": "[^"]+", "body": "[^"]+"}}' } def validate_tool_call(llm_output: str) -> dict: for tool, pattern in TOOL_CALL_TEMPLATES.items(): if re.fullmatch(pattern, llm_output.strip()): return json.loads(llm_output.strip()) raise ValueError("Invalid tool call format")
  2. 参数沙箱:所有工具参数必须通过白名单校验

    def sanitize_email_params(params: dict)

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

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

立即咨询