2026年Q1季度，我们团队在HolySheep平台上承接了17个企业级Agent项目，总对话轮次超过4200万次。在这些项目中，我亲历了9次严重的生产事故，其中包括一次因工具调用链断裂导致3小时服务中断的事件，另一次因为模型重试策略不当造成日均$2,300的预算在4小时内耗尽。这些惨痛的教训让我深刻认识到：一个完善的Agent生产事故复盘模板，比任何监控告警都重要。
本文将从工具调用链追踪、模型重试策略、人工接管节点设计三个维度，详细讲解我在HolySheep平台上构建的完整复盘体系。每一个章节都附带可直接上线的生产级代码，以及我踩坑后的真实Benchmark数据。
一、工具调用链追踪：构建可观测的Agent执行图谱
在传统Agent架构中，工具调用链往往是黑盒状态——你只知道最终结果，不知道中间经历了什么。2025年双十一期间，我们的客服Agent因为API超时导致整个对话卡死，最后排查了6小时才发现是某个第三方天气API响应时间从200ms飙升至28秒。这个案例让我决定在HolySheep平台上构建完整的调用链追踪系统。
HolySheep API支持流式输出的同时返回完整的使用量统计，这为我们构建调用链追踪提供了基础能力。通过在每次工具调用时记录上下文快照，我们可以精确还原Agent的决策路径。
1.1 调用链数据结构设计
我在项目中设计了四层追踪数据结构，从宏观到微观完整覆盖Agent执行过程：
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, List, Dict, Any
from enum import Enum
import asyncio
import json
class NodeStatus(Enum):
PENDING = "pending"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
RETRYING = "retrying"
MANUAL_ESCALATION = "manual_escalation"
class ToolType(Enum):
SEARCH = "search"
CALCULATOR = "calculator"
API_CALL = "api_call"
DATABASE = "database"
FILE_SYSTEM = "file_system"
HUMAN_REVIEW = "human_review"
@dataclass
class ToolCallNode:
"""单个工具调用节点"""
node_id: str
tool_name: str
tool_type: ToolType
input_params: Dict[str, Any]
output_result: Optional[Any] = None
error_message: Optional[str] = None
status: NodeStatus = NodeStatus.PENDING
start_time: Optional[datetime] = None
end_time: Optional[datetime] = None
retry_count: int = 0
max_retries: int = 3
parent_node_id: Optional[str] = None
children_node_ids: List[str] = field(default_factory=list)
latency_ms: float = 0.0
cost_usd: float = 0.0
model_name: Optional[str] = None
def __post_init__(self):
if self.start_time is None:
self.start_time = datetime.utcnow()
@property
def duration_ms(self) -> float:
if self.end_time and self.start_time:
return (self.end_time - self.start_time).total_seconds() * 1000
return 0.0
def to_dict(self) -> Dict[str, Any]:
return {
"node_id": self.node_id,
"tool_name": self.tool_name,
"tool_type": self.tool_type.value,
"input_params": self.input_params,
"output_result": str(self.output_result)[:500] if self.output_result else None,
"error_message": self.error_message,
"status": self.status.value,
"start_time": self.start_time.isoformat() if self.start_time else None,
"end_time": self.end_time.isoformat() if self.end_time else None,
"retry_count": self.retry_count,
"latency_ms": self.latency_ms,
"cost_usd": self.cost_usd,
"duration_ms": self.duration_ms
}
@dataclass
class ExecutionTrace:
"""完整执行追踪"""
trace_id: str
session_id: str
user_query: str
nodes: List[ToolCallNode] = field(default_factory=list)
created_at: datetime = field(default_factory=datetime.utcnow)
total_latency_ms: float = 0.0
total_cost_usd: float = 0.0
total_tokens: int = 0
final_status: NodeStatus = NodeStatus.PENDING
escalation_triggered: bool = False
escalation_reason: Optional[str] = None
def add_node(self, node: ToolCallNode) -> None:
"""添加节点并更新父子关系"""
self.nodes.append(node)
if node.parent_node_id:
for parent in self.nodes:
if parent.node_id == node.parent_node_id:
parent.children_node_ids.append(node.node_id)
def calculate_totals(self) -> None:
"""计算汇总指标"""
self.total_latency_ms = sum(n.latency_ms for n in self.nodes)
self.total_cost_usd = sum(n.cost_usd for n in self.nodes)
self.total_tokens = sum(n.latency_ms for n in self.nodes) # 简化示例
def get_critical_path(self) -> List[ToolCallNode]:
"""获取关键路径（耗时最长的节点链）"""
sorted_nodes = sorted(self.nodes, key=lambda x: x.latency_ms, reverse=True)
return sorted_nodes[:5] # 返回Top5耗时节点
def to_incident_report(self) -> Dict[str, Any]:
"""生成事故报告"""
self.calculate_totals()
return {
"trace_id": self.trace_id,
"session_id": self.session_id,
"user_query": self.user_query,
"total_nodes": len(self.nodes),
"total_latency_ms": self.total_latency_ms,
"total_cost_usd": self.total_cost_usd,
"final_status": self.final_status.value,
"escalation_triggered": self.escalation_triggered,
"critical_path": [n.tool_name for n in self.get_critical_path()],
"failed_nodes": [n.node_id for n in self.nodes if n.status == NodeStatus.FAILED],
"report_generated_at": datetime.utcnow().isoformat()
}1.2 与HolySheep API集成实现毫秒级追踪
下面是与HolySheep API深度集成的完整实现，实现了从模型调用到工具执行的端到端追踪：
import aiohttp
import asyncio
import uuid
from typing import List, Dict, Any, Callable, Optional
from datetime import datetime
import json
import hashlib
class HolySheepAgentTracker:
"""HolySheep平台Agent追踪器 - 生产级别实现"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(
self,
api_key: str,
max_concurrent_calls: int = 10,
timeout_seconds: int = 30,
enable_auto_retry: bool = True,
enable_manual_escalation: bool = True
):
self.api_key = api_key
self.max_concurrent_calls = max_concurrent_calls
self.timeout_seconds = timeout_seconds
self.enable_auto_retry = enable_auto_retry
self.enable_manual_escalation = enable_manual_escalation
self.semaphore = asyncio.Semaphore(max_concurrent_calls)
# 追踪存储
self.active_traces: Dict[str, ExecutionTrace] = {}
self.completed_traces: List[ExecutionTrace] = []
# 性能指标
self.metrics = {
"total_calls": 0,
"failed_calls": 0,
"retried_calls": 0,
"escalated_calls": 0,
"avg_latency_ms": 0.0,
"total_cost_usd": 0.0
}
def _generate_node_id(self, prefix: str = "node") -> str:
"""生成唯一节点ID"""
timestamp = datetime.utcnow().isoformat()
raw = f"{prefix}_{timestamp}_{uuid.uuid4().hex[:8]}"
return hashlib.md5(raw.encode()).hexdigest()[:16]
async def call_with_tracking(
self,
trace: ExecutionTrace,
tool_func: Callable,
tool_name: str,
tool_type: ToolType,
parent_node_id: Optional[str] = None,
**kwargs
) -> ToolCallNode:
"""带追踪的模型调用"""
node = ToolCallNode(
node_id=self._generate_node_id(tool_name),
tool_name=tool_name,
tool_type=tool_type,
input_params=kwargs,
parent_node_id=parent_node_id,
max_retries=kwargs.get('max_retries', 3)
)
trace.add_node(node)
node.start_time = datetime.utcnow()
try:
async with self.semaphore: # 并发控制
node.status = NodeStatus.RUNNING
# 执行工具调用
start_ts = asyncio.get_event_loop().time()
result = await asyncio.wait_for(
tool_func(**kwargs),
timeout=self.timeout_seconds
)
end_ts = asyncio.get_event_loop().time()
node.latency_ms = (end_ts - start_ts) * 1000
node.output_result = result
node.status = NodeStatus.SUCCESS
self.metrics["total_calls"] += 1
self.metrics["total_cost_usd"] += node.cost_usd
except asyncio.TimeoutError:
node.error_message = f"Timeout after {self.timeout_seconds}s"
node.status = NodeStatus.FAILED
if self.enable_auto_retry and node.retry_count < node.max_retries:
node = await self._retry_node(trace, node, tool_func, **kwargs)
except Exception as e:
node.error_message = str(e)
node.status = NodeStatus.FAILED
if self.enable_manual_escalation:
node = await self._handle_escalation(trace, node, str(e))
node.end_time = datetime.utcnow()
return node
async def _retry_node(
self,
trace: ExecutionTrace,
node: ToolCallNode,
tool_func: Callable,
**kwargs
) -> ToolCallNode:
"""指数退避重试"""
node.status = NodeStatus.RETRYING
node.retry_count += 1
self.metrics["retried_calls"] += 1
# 指数退避: 1s, 2s, 4s
backoff = min(2 ** (node.retry_count - 1), 10)
await asyncio.sleep(backoff)
# 重新执行
node.start_time = datetime.utcnow()
try:
result = await asyncio.wait_for(
tool_func(**kwargs),
timeout=self.timeout_seconds
)
node.output_result = result
node.status = NodeStatus.SUCCESS
node.end_time = datetime.utcnow()
except Exception as e:
node.error_message = f"Retry {node.retry_count} failed: {str(e)}"
node.status = NodeStatus.FAILED
node.end_time = datetime.utcnow()
return node
async def _handle_escalation(
self,
trace: ExecutionTrace,
node: ToolCallNode,
error: str
) -> ToolCallNode:
"""人工接管处理"""
node.status = NodeStatus.MANUAL_ESCALATION
trace.escalation_triggered = True
trace.escalation_reason = error
self.metrics["escalated_calls"] += 1
# 创建人工审核节点
review_node = ToolCallNode(
node_id=self._generate_node_id("human_review"),
tool_name="human_review_required",
tool_type=ToolType.HUMAN_REVIEW,
input_params={
"original_error": error,
"failed_node_id": node.node_id,
"user_query": trace.user_query
},
parent_node_id=node.node_id,
status=NodeStatus.MANUAL_ESCALATION
)
trace.add_node(review_node)
return node
async def chat_completion_with_tools(
self,
messages: List[Dict[str, str]],
tools: List[Dict[str, Any]],
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict[str, Any]:
"""HolySheep API完整调用示例"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"tools": tools,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": False
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
if response.status != 200:
error_text = await response.text()
raise Exception(f"HolySheep API Error {response.status}: {error_text}")
result = await response.json()
return {
"content": result["choices"][0]["message"]["content"],
"tool_calls": result["choices"][0]["message"].get("tool_calls", []),
"usage": result.get("usage", {}),
"model": result.get("model"),
"response_id": result.get("id")
}
def generate_incident_markdown(self, trace: ExecutionTrace) -> str:
"""生成事故复盘Markdown"""
report = trace.to_incident_report()
md = f"""# 🚨 Agent生产事故复盘报告
基本信息
- **Trace ID**: {report['trace_id']}
- **Session ID**: {report['session_id']}
- **发生时间**: {trace.created_at.strftime('%Y-%m-%d %H:%M:%S')}
- **用户Query**: {report['user_query']}
执行摘要
| 指标 | 数值 |
|------|------|
| 总调用节点数 | {report['total_nodes']} |
| 总耗时 | {report['total_latency_ms']:.2f}ms |
| 总成本 | ${report['total_cost_usd']:.4f} |
| 最终状态 | {report['final_status']} |
| 是否人工接管 | {'是' if report['escalation_triggered'] else '否'} |
关键路径分析（Top 5耗时节点）
{', '.join(report['critical_path'])}
失败节点
{', '.join(report['failed_nodes']) if report['failed_nodes'] else '无'}
根因分析
> 待填写
改进措施
- [ ] 待填写
复盘结论
> 待填写
"""
return md
使用示例
async def main():
tracker = HolySheepAgentTracker(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent_calls=10,
enable_auto_retry=True
)
# 创建追踪实例
trace = ExecutionTrace(
trace_id=uuid.uuid4().hex,
session_id="sess_123456",
user_query="帮我查询北京明天的天气并安排行程"
)
# 定义工具
async def weather_tool(city: str, date: str) -> str:
await asyncio.sleep(0.5) # 模拟API调用
return f"{city} {date} 晴转多云 15-25°C"
# 执行带追踪的调用
result = await tracker.call_with_tracking(
trace=trace,
tool_func=weather_tool,
tool_name="weather查询",
tool_type=ToolType.API_CALL,
city="北京",
date="明天"
)
print(f"调用成功，延迟: {result.latency_ms:.2f}ms")
print(tracker.generate_incident_markdown(trace))
if __name__ == "__main__":
asyncio.run(main())1.3 生产Benchmark数据
经过3个月的生产验证，我们积累了以下性能数据（基于日均100万次调用的集群）：
| 指标 | 优化前 | 优化后 | 提升幅度 |
|---|---|---|---|
| 平均追踪延迟 | 45ms | 8ms | +82% |
| 调用链完整率 | 67.3% | 99.7% | +48% |
| 事故平均定位时间 | 4.2小时 | 12分钟 | +95% |
| 重复事故发生率 | 23.5% | 2.1% | +91% |
| 月均API成本 | $18,500 | $12,300 | -33% |
二、模型重试策略：避免预算在4小时内燃尽
2025年12月的一个深夜，我被PagerDuty的告警叫醒：当日API账单已达$9,800，而当时才凌晨2点。排查后发现，是因为某个Agent的错误处理逻辑触发了死循环重试，单个会话在1小时内产生了47,000次API调用。这个血的教训让我重新设计了整套重试策略。
2.1 智能重试策略实现
from typing import TypeVar, Callable, Optional, Tuple, List
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
import asyncio
import random
import json
import hashlib
T = TypeVar('T')
class RetryStrategy(Enum):
EXPONENTIAL_BACKOFF = "exponential_backoff"
LINEAR_BACKOFF = "linear_backoff"
FIBONACCI_BACKOFF = "fibonacci_backoff"
IMMEDIATE = "immediate"
@dataclass
class RetryConfig:
"""重试配置"""
max_retries: int = 3
base_delay_seconds: float = 1.0
max_delay_seconds: float = 60.0
strategy: RetryStrategy = RetryStrategy.EXPONENTIAL_BACKOFF
jitter: bool = True
jitter_factor: float = 0.3
# 预算保护
max_budget_per_call_usd: float = 0.50
max_total_budget_usd: float = 100.0
current_budget_usd: float = 0.0
# 熔断器
failure_threshold: int = 5
failure_window_seconds: int = 300
recovery_timeout_seconds: int = 600
class CircuitBreaker:
"""熔断器实现"""
def __init__(self, config: RetryConfig):
self.config = config
self.failures: List[datetime] = []
self.state = "closed" # closed, open, half_open
self.last_failure_time: Optional[datetime] = None
def record_failure(self) -> None:
self.failures.append(datetime.utcnow())
self.last_failure_time = datetime.utcnow()
# 清理过期记录
cutoff = datetime.utcnow() - timedelta(seconds=self.config.failure_window_seconds)
self.failures = [f for f in self.failures if f > cutoff]
if len(self.failures) >= self.config.failure_threshold:
self.state = "open"
def record_success(self) -> None:
self.state = "closed"
self.failures = []
def can_execute(self) -> bool:
if self.state == "closed":
return True
if self.state == "open":
if self.last_failure_time:
elapsed = (datetime.utcnow() - self.last_failure_time).total_seconds()
if elapsed >= self.config.recovery_timeout_seconds:
self.state = "half_open"
return True
return False
return True # half_open
class SmartRetryExecutor:
"""智能重试执行器"""
def __init__(self, api_key: str, config: Optional[RetryConfig] = None):
self.config = config or RetryConfig()
self.circuit_breaker = CircuitBreaker(self.config)
self.api_key = api_key
self.call_history: List[Dict] = []
def _calculate_delay(self, attempt: int) -> float:
"""计算退避延迟"""
if self.config.strategy == RetryStrategy.EXPONENTIAL_BACKOFF:
delay = self.config.base_delay_seconds * (2 ** attempt)
elif self.config.strategy == RetryStrategy.LINEAR_BACKOFF:
delay = self.config.base_delay_seconds * (attempt + 1)
elif self.config.strategy == RetryStrategy.FIBONACCI_BACKOFF:
delay = self.config.base_delay_seconds * self._fibonacci(attempt + 1)
else:
delay = 0
delay = min(delay, self.config.max_delay_seconds)
if self.config.jitter:
jitter_range = delay * self.config.jitter_factor
delay += random.uniform(-jitter_range, jitter_range)
return max(0, delay)
def _fibonacci(self, n: int) -> int:
if n <= 1:
return 1
a, b = 1, 1
for _ in range(n - 1):
a, b = b, a + b
return b
def _should_retry(self, attempt: int, error: Exception, result: Optional[T]) -> Tuple[bool, str]:
"""判断是否应该重试"""
# 预算耗尽检查
if self.config.current_budget_usd >= self.config.max_total_budget_usd:
return False, "Budget exhausted"
# HTTP状态码判断
error_str = str(error)
if "429" in error_str:
return True, "Rate limited - retry with backoff"
if "500" in error_str or "502" in error_str or "503" in error_str:
return True, "Server error - retry safe"
if "timeout" in error_str.lower():
return True, "Timeout - retry safe"
if "connection" in error_str.lower():
return True, "Connection error - retry safe"
# 4xx错误通常不重试
if any(code in error_str for code in ["400", "401", "403", "404"]):
return False, "Client error - no retry"
return attempt < self.config.max_retries, f"Attempt {attempt + 1} of {self.config.max_retries}"
async def execute_with_retry(
self,
func: Callable,
*args,
call_cost_usd: float = 0.0,
**kwargs
) -> Tuple[T, Dict]:
"""
执行带重试的调用
Returns:
(result, metadata) - 结果和元数据
"""
attempt = 0
last_error = None
start_time = datetime.utcnow()
metadata = {
"attempts": 0,
"success": False,
"total_cost_usd": 0.0,
"total_latency_ms": 0.0,
"errors": []
}
while True:
if not self.circuit_breaker.can_execute():
raise Exception("Circuit breaker is open - service unavailable")
metadata["attempts"] = attempt + 1
# 预算预检查
if call_cost_usd > self.config.max_budget_per_call_usd:
raise Exception(f"Per-call budget exceeded: ${call_cost_usd:.4f} > ${self.config.max_budget_per_call_usd}")
try:
call_start = asyncio.get_event_loop().time()
result = await func(*args, **kwargs)
call_end = asyncio.get_event_loop().time()
latency_ms = (call_end - call_start) * 1000
metadata["total_latency_ms"] += latency_ms
metadata["total_cost_usd"] += call_cost_usd
metadata["success"] = True
self.circuit_breaker.record_success()
self.call_history.append({
"timestamp": datetime.utcnow(),
"success": True,
"latency_ms": latency_ms,
"cost_usd": call_cost_usd,
"attempt": attempt
})
return result, metadata
except Exception as e:
last_error = e
metadata["errors"].append(str(e))
call_end = asyncio.get_event_loop().time()
latency_ms = (call_end - call_start) * 1000
metadata["total_latency_ms"] += latency_ms
self.circuit_breaker.record_failure()
should_retry, reason = self._should_retry(attempt, e, None)
if not should_retry:
raise Exception(f"Retry exhausted: {reason}. Last error: {last_error}")
attempt += 1
if attempt <= self.config.max_retries:
delay = self._calculate_delay(attempt - 1)
await asyncio.sleep(delay)
else:
break
raise Exception(f"Max retries exceeded. Errors: {metadata['errors']}")
def get_health_report(self) -> Dict:
"""获取健康报告"""
recent_calls = [
c for c in self.call_history
if c["timestamp"] > datetime.utcnow() - timedelta(minutes=5)
]
if not recent_calls:
return {"status": "no_data"}
success_rate = sum(1 for c in recent_calls if c["success"]) / len(recent_calls)
avg_latency = sum(c["latency_ms"] for c in recent_calls) / len(recent_calls)
return {
"status": self.circuit_breaker.state,
"recent_calls": len(recent_calls),
"success_rate": f"{success_rate * 100:.2f}%",
"avg_latency_ms": f"{avg_latency:.2f}",
"failure_count": len(self.circuit_breaker.failures)
}
HolySheep API实际调用示例
async def call_holysheep_with_retry(executor: SmartRetryExecutor, prompt: str):
"""使用重试策略调用HolySheep API"""
headers = {
"Authorization": f"Bearer {executor.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "claude-sonnet-4.5",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1024
}
async def _call():
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload
) as resp:
if resp.status != 200:
raise Exception(f"API Error {resp.status}: {await resp.text()}")
return await resp.json()
# 估算成本（实际根据返回的usage计算）
estimated_cost = 0.00015 # $0.00015 per call (Sonnet 4.5 output)
result, metadata = await executor.execute_with_retry(
_call,
call_cost_usd=estimated_cost
)
print(f"调用成功，尝试次数: {metadata['attempts']}, 总延迟: {metadata['total_latency_ms']:.2f}ms")
return result2.2 预算保护机制实战
我遇到过最严重的一次事故是：凌晨3点，一个对话式搜索Agent陷入了死循环，4小时内产生了14万次API调用，账单高达$4,200。事后我设计了这套预算保护机制，确保类似情况不会再次发生。
| 保护层级 | 阈值设置 | 触发动作 | 恢复方式 |
|---|---|---|---|
| L1 单次调用 | $0.50 | 拒绝执行 | 人工审核 |
| L2 会话累计 | $5.00 | 暂停API调用 | 冷却30分钟 |
| L3 小时预算 | $50.00 | 熔断器开启 | 手动关闭 |
| L4 日预算 | $500.00 | 全部服务暂停 | 联系支持 |
三、人工接管节点设计：让Agent学会说"我不知道"
2026年2月，我们的金融分析Agent在用户询问"某上市公司是否会被收购"时，给出了一个看似专业但实际上是胡编乱造的答案。这个案例让我意识到：不是所有问题都应该让AI强答，有时候说"我不知道，请人工介入"才是最好的答案。
3.1 人工接管触发器实现
from typing import Optional, Dict, Any, List, Callable
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
import asyncio
import re
class EscalationTrigger(Enum):
LOW_CONFIDENCE = "low_confidence"
SENSITIVE_TOPIC = "sensitive_topic"
TOOL_FAILURE = "tool_failure"
USER_EXPLICIT = "user_explicit"
SAFETY_FLAG = "safety_flag"
CONFLICTING_INFO = "conflicting_info"
@dataclass
class EscalationRule:
"""人工接管规则"""
trigger: EscalationTrigger
pattern: Optional[str] = None
keywords: Optional[List[str]] = None
confidence_threshold: float = 0.6
priority: int = 1
auto_escalate: bool = True
class HumanEscalationManager:
"""人工接管管理器"""
# 敏感话题关键词库
SENSITIVE_KEYWORDS = [
"投资建议", "股票推荐", "内幕消息", "财务预测",
"法律诉讼", "犯罪行为", "政治敏感", "医疗建议",
"处方药", "自杀", "暴力", "儿童安全"
]
# 触发规则
DEFAULT_RULES = [
EscalationRule(
trigger=EscalationTrigger.LOW_CONFIDENCE,
confidence_threshold=0.65
),
EscalationRule(
trigger=EscalationTrigger.SENSITIVE_TOPIC,
keywords=SENSITIVE_KEYWORDS,
priority=1
),
EscalationRule(
trigger=EscalationTrigger.SAFETY_FLAG,
pattern=r"(?:自杀|自残|如何杀人|炸弹制作)"
),
EscalationRule(
trigger=EscalationTrigger.USER_EXPLICIT,
keywords=["人工服务", "转人工", "人工客服", "找真人"]
)
]
def __init__(
self,
escalation_callback: Optional[Callable] = None,
webhook_url: Optional[str] = None,
slack_channel: Optional[str] = None,
enable_pagerduty: bool = False
):
self.rules = self.DEFAULT_RULES.copy()
self.escalation_callback = escalation_callback
self.webhook_url = webhook_url
self.slack_channel = slack_channel
self.enable_pagerduty = enable_pagerduty
self.pending_escalations: Dict[str, Dict] = {}
self.escalation_history: List[Dict] = []
def add_rule(self, rule: EscalationRule) -> None:
self.rules.append(rule)
self.rules.sort(key=lambda x: x.priority, reverse=True)
def check_escalation(
self,
user_query: str,
model_response: str,
confidence_score: float,
tool_results: Optional[List[Dict]] = None,
context: Optional[Dict] = None
) -> Optional[Dict]:
"""检查是否需要人工接管"""
escalation = None
for rule in self.rules:
should_escalate = False
reason = ""
if rule.trigger == EscalationTrigger.LOW_CONFIDENCE:
if confidence_score < rule.confidence_threshold:
should_escalate = True
reason = f"低置信度: {confidence_score:.2f} < {rule.confidence_threshold}"
elif rule.trigger == EscalationTrigger.SENSITIVE_TOPIC:
for keyword in rule.keywords or []:
if keyword in user_query or keyword in model_response:
should_escalate = True
reason = f"敏感话题: {keyword}"
break
elif rule.trigger == EscalationTrigger.SAFETY_FLAG:
if rule.pattern and re.search(rule.pattern, user_query):
should_escalate = True
reason = "安全关键词触发"
elif rule.trigger == EscalationTrigger.USER_EXPLICIT:
for keyword in rule.keywords or []:
if keyword in user_query:
should_escalate = True
reason = f"用户明确请求: {keyword}"
break
elif rule.trigger == EscalationTrigger.TOOL_FAILURE:
if tool_results and any(t.get("error") for t in tool_results):
failed_tools = [t["tool"] for t in tool_results if t.get("error")]
should_escalate = True
reason = f"工具调用失败: {', '.join(failed_tools)}"
elif rule.trigger == EscalationTrigger.CONFLICTING_INFO:
if tool_results and len(tool_results) > 1:
# 检测信息冲突
results_text = [str(t.get("result", "")) for t in tool_results]
if self._has_conflict(results_text):
should_escalate = True
reason = "多个数据源信息冲突"
if should_escalate:
escalation = {
"trigger": rule.trigger.value,
"reason": reason,
"priority": rule.priority,
"auto_escalate": rule.auto_escalate,
"timestamp": datetime.utcnow().isoformat(),
"user_query": user_query[:200],
"model_response": model_response[:500] if model_response else None,
"confidence_score": confidence