当前位置：首页 > news >正文

Prompt Engineering 与 Agent 工作流构建：典型线上故障的定位证据链

news 2026/8/10 21:37:46

Prompt Engineering 与 Agent 工作流构建：典型线上故障的定位证据链

本文围绕“典型线上故障的定位证据链”梳理可执行的工程取舍与检查重点。文中的配置、阈值和示例用于说明设计方法；接入实际项目时，应根据业务场景、监控数据和依赖能力完成验证。

许多人遇到这种故障，第一反应是改写 Prompt，加一句“请务必返回合法的 JSON 格式”。但在复杂的 LLM Agent 工作流里，光靠口头告诫模型往往无济于事。如果没有一套完整的线上故障定位证据链，一次失败的实验只会沦为相互猜忌的口水仗。

当模型在第 4 轮调用中突然丢掉闭合括号

问题很少发生在第一轮简单对话中。系统刚启动时，Prompt 简短，上下文干爽，模型表现得宛如天才。当工作流推进到第 4 轮甚至第 7 轮，累积的工具返回结果、遗留的思考链（Chain-of-Thought）以及逐步膨胀的 Token 计数，会像滚雪球一样蚕食模型的注意力和上下文窗口。

在这次失败的实验中，通过抓取原始 Payload 发现，模型并非没有理解指令，而是当输入 Token 突破 6000 临界点时，为了强行满足输出格式要求，模型在生成 Markdown 代码块包裹的 JSON 字符串时，尾部的}被截断或混入了非标准换行符。下游逻辑直接崩溃在json.loads()这一行。

单纯看 API 调用的 HTTP 状态码，系统返回的是标准的 200 OK。只有把单次 Agent 运行过程中的输入上下文、Prompt 模版演变、中间工具调用记录以及底层原始 Stream 片段完整串联起来，故障发生的因果推演才清晰可见。

构建具备因果链的日志结构与上下文状态快照

排查 Agent 工作流的故障，需要将离散的 HTTP 请求转变为连续的因果证据链。传统的日志打印（如logger.info("Calling LLM...")）在并发场景下极易被打碎，无法还原特定 Session 的思维演进轨迹。

一套健全的 Agent 证据链系统，必须包含以下核心要素：

TraceID 与 ExecutionStep 的双重锚定：为每一次 Agent 任务生成唯一追踪 ID，并在多轮思考与工具调用中递增步骤索引。
Context Snapshot（上下文快照）：在发送给大模型之前，冻结并记录当前历史对话队列、Token 消耗预估及系统 Prompt 版本号。
Raw Output Capture（原始输出冻结）：即使下游 Schema 校验失败，也必须保留大模型未经任何清洗的真实字节流。

flowchart TD A[用户请求输入] --> B[生成 Agent TraceID] B --> C[装配 Prompt 与历史 Context] C --> D[创建 StateSnapshot 现场快照] D --> E[调用 LLM 模型接口] E -->|成功返回| F[提取原始 Stream/Text] E -->|超时或异常| G[捕获 TransportError 并挂起现场] F --> H{Schema 解析与 JSON 验证} H -->|解析成功| I[更新 Session 状态并执行工具] H -->|解析失败| J[记录 RawOutput 到 ErrorBucket] J --> K[触发级联降级策略/带退避重试] G --> K I --> L[输出下一步思考或最终响应]

带级联重试与异常现场冻结的 Agent 追踪器实现

为了在生产环境捕获完整的证据链，我们需要编写一个高容错、具备上下文快照与结构化归档功能的 Agent 执行器。以下代码演示了如何在 Python 中构建这套防护网：

import json import logging import time import uuid from typing import Dict, Any, List, Optional, Callable from dataclasses import dataclass, field, asdict # 配置结构化日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - [%(levelname)s] - %(message)s') logger = logging.getLogger("AgentTraceLogger") @dataclass class ExecutionStep: step_id: str step_index: int prompt_version: str messages_snapshot: List[Dict[str, str]] raw_response: Optional[str] = None parsed_output: Optional[Dict[str, Any]] = None error_message: Optional[str] = None duration_ms: float = 0.0 @dataclass class AgentTraceContext: trace_id: str session_id: str created_at: float = field(default_factory=time.time) steps: List[ExecutionStep] = field(default_factory=list) def to_json(self) -> str: return json.dumps(asdict(self), ensure_ascii=False, indent=2) class SchemaValidationError(Exception): """当模型输出无法通过 Schema 解析时抛出""" def __init__(self, message: str, raw_output: str): super().__init__(message) self.raw_output = raw_output class ResilientAgentExecutor: def __init__(self, prompt_version: str = "v1.2.0", max_retries: int = 3): self.prompt_version = prompt_version self.max_retries = max_retries def execute_step( self, context: AgentTraceContext, history_messages: List[Dict[str, str]], mock_llm_call: Callable[[List[Dict[str, str]]], str] ) -> Dict[str, Any]: step_idx = len(context.steps) + 1 step_id = f"step-{uuid.uuid4().hex[:8]}" start_time = time.time() # 1. 冻结输入上下文快照 snapshot_messages = [msg.copy() for msg in history_messages] current_step = ExecutionStep( step_id=step_id, step_index=step_idx, prompt_version=self.prompt_version, messages_snapshot=snapshot_messages ) raw_output = "" attempt = 0 while attempt < self.max_retries: attempt += 1 try: logger.info(f"[Trace: {context.trace_id}] 执行步骤 {step_idx}，第 {attempt} 次尝试") # 2. 调用大模型（包含超时防护） raw_output = mock_llm_call(history_messages) current_step.raw_response = raw_output # 3. 解析与严格 Schema 验证 parsed = self._parse_and_validate(raw_output) current_step.parsed_output = parsed current_step.duration_ms = (time.time() - start_time) * 1000 context.steps.append(current_step) return parsed except (json.JSONDecodeError, SchemaValidationError) as val_err: logger.warning(f"[Trace: {context.trace_id}] Schema 解析失败: {str(val_err)}") current_step.error_message = f"Attempt {attempt} failed: {str(val_err)}" if attempt >= self.max_retries: current_step.duration_ms = (time.time() - start_time) * 1000 context.steps.append(current_step) # 归档异常快照，供后续复盘分析 self._archive_trace(context, is_fatal=True) raise SchemaValidationError( f"连续 {self.max_retries} 次解析失败，Agent 工作流终止", raw_output=raw_output ) # 指数退避等待，避开偶发性响应畸变 time.sleep(0.5 * (2 ** (attempt - 1))) except Exception as ex: logger.error(f"[Trace: {context.trace_id}] 未知运行时异常: {str(ex)}") current_step.error_message = f"Fatal Runtime Exception: {str(ex)}" current_step.duration_ms = (time.time() - start_time) * 1000 context.steps.append(current_step) self._archive_trace(context, is_fatal=True) raise ex raise RuntimeError("意外出界执行路径") def _parse_and_validate(self, raw_text: str) -> Dict[str, Any]: """清洗 Markdown 标识并验证 JSON 格式""" cleaned = raw_text.strip() if cleaned.startswith("```json"): cleaned = cleaned[7:] if cleaned.startswith("```"): cleaned = cleaned[3:] if cleaned.endswith("```"): cleaned = cleaned[:-3] cleaned = cleaned.strip() if not cleaned: raise SchemaValidationError("模型返回内容为空", raw_text) try: data = json.loads(cleaned) except json.JSONDecodeError as e: raise SchemaValidationError(f"JSON 语法损坏: {e.msg}", raw_text) # 检查业务必需字段 if "action" not in data or "thought" not in data: raise SchemaValidationError("缺少必需业务字段 'action' 或 'thought'", raw_text) return data def _archive_trace(self, context: AgentTraceContext, is_fatal: bool = False): prefix = "FATAL" if is_fatal else "INFO" logger.info(f"[{prefix} ARCHIVE] 正在序列化 Agent 证据链日志...\n{context.to_json()}") # 模拟验证演示 if __name__ == "__main__": trace_ctx = AgentTraceContext(trace_id="tr-889012", session_id="sess-9901") executor = ResilientAgentExecutor(prompt_version="v2.1-bugfix", max_retries=2) # 模拟一个会返回残缺 JSON 的 LLM 函数 def faulty_llm_provider(messages: List[Dict[str, str]]) -> str: # 模拟第 4 轮出现的坏帧 return '```json\n{"thought": "需要查询用户数据", "action": "query_db"' # 缺少闭合括号 try: sample_messages = [{"role": "user", "content": "请分析用户 1002 的订单趋势"}] executor.execute_step(trace_ctx, sample_messages, faulty_llm_provider) except SchemaValidationError as e: print(f"\n捕获到预期异常！原始故障 payload 已冻结。异常消息: {e}")