分析智能体故障的根本原因
系统化故障分类：模型错误、工具错误、数据错误、逻辑错误
分析智能体故障的根本原因 是 CoddyKit 上的免费 AI Agents 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。
代理故障分类
代理故障分为四类：
- 模型错误：LLM 调用了错误的工具或生成了错误的输出
- 工具错误：外部 API 失败或返回意外数据
- 数据错误：输入错误（格式错误、字段缺失、类型意外）
- 逻辑错误：步骤正确，但顺序错误或假设错误
模型错误：调用了错误的工具
模型错误发生在 LLM 选择了错误的工具、传入了不正确的参数，或生成格式错误的 JSON 时。这些错误通常由不清晰的工具说明或含义模糊的提示词导致。
import openai
import json
client = openai.OpenAI(api_key='sk-...')
def detect_model_errors(response) -> list:
errors = []
message = response.choices[0].message
if message.tool_calls:
for tc in message.tool_calls:
tool_name = tc.function.name
try:
args = json.loads(tc.function.arguments)
except json.JSONDecodeError as e:
errors.append({
'type': 'model_error',
'subtype': 'malformed_tool_args',
'tool': tool_name,
'raw_args': tc.function.arguments,
'parse_error': str(e)
})
continue
# Validate required arguments
expected_tools = {
'search_web': ['query'],
'send_email': ['to', 'subject', 'body'],
'create_task': ['title']
}
required = expected_tools.get(tool_name, [])
missing = [r for r in required if r not in args]
if missing:
errors.append({
'type': 'model_error',
'subtype': 'missing_required_args',
'tool': tool_name,
'missing': missing
})
return errors
print('Model error detection function defined')工具错误：API 失败
工具错误发生在外部 API 返回错误（4xx、5xx）、超时，或以意外格式返回数据时。请捕获确切错误、工具参数和响应，以便诊断。
import httpx
from dataclasses import dataclass
from typing import Optional
@dataclass
class ToolError:
tool_name: str
error_type: str
status_code: Optional[int]
message: str
args_used: dict
retry_possible: bool
def classify_tool_error(tool_name: str, args: dict, exception: Exception) -> ToolError:
if isinstance(exception, httpx.TimeoutException):
return ToolError(
tool_name=tool_name,
error_type='timeout',
status_code=None,
message=str(exception),
args_used=args,
retry_possible=True # Retries are appropriate for timeouts
)
elif isinstance(exception, httpx.HTTPStatusError):
status = exception.response.status_code
retry = status >= 500 or status == 429 # Server errors and rate limits are retryable
return ToolError(
tool_name=tool_name,
error_type='http_error',
status_code=status,
message=exception.response.text[:200],
args_used=args,
retry_possible=retry
)
else:
return ToolError(
tool_name=tool_name,
error_type='unexpected_error',
status_code=None,
message=str(exception),
args_used=args,
retry_possible=False
)
print('Tool error classification defined')数据错误：输入验证
数据错误源于传给代理的错误输入：必填字段缺失、数据类型错误，或在应为数字的位置传入字符串。请在代理入口验证输入，以便及早捕获这些错误。
from pydantic import BaseModel, validator, ValidationError
from typing import Optional
class EmailAgentInput(BaseModel):
email_id: str
action: str
user_id: int
priority: Optional[str] = 'normal'
@validator('action')
def action_must_be_valid(cls, v):
valid_actions = ['reply', 'forward', 'archive', 'summarize']
if v not in valid_actions:
raise ValueError(f'action must be one of {valid_actions}, got: {v}')
return v
@validator('email_id')
def email_id_not_empty(cls, v):
if not v.strip():
raise ValueError('email_id cannot be empty')
return v
def validate_agent_input(raw_input: dict) -> tuple:
try:
validated = EmailAgentInput(**raw_input)
return validated, None
except ValidationError as e:
return None, [
{'field': err['loc'][0], 'message': err['msg']}
for err in e.errors()
]
# Test with bad input
valid, errors = validate_agent_input({'email_id': '', 'action': 'delete', 'user_id': 'abc'})
if errors:
print('Data errors found:')
for err in errors:
print(f' {err["field"]}: {err["message"]}')逻辑错误：步骤顺序错误
逻辑错误最难调试。代理使用正确的参数调用正确的工具，但顺序错误、跳过了必需步骤，或对先前步骤的输出作出了错误假设。
import logging
logger = logging.getLogger('agent.logic')
class AgentStepGuard:
'''
Enforces that steps execute in the required sequence.
'''
def __init__(self):
self.completed_steps = set()
self.STEP_DEPENDENCIES = {
'extract_action_items': ['read_email'],
'create_trello_card': ['extract_action_items'],
'send_slack_notification': ['create_trello_card']
}
def mark_complete(self, step_name: str):
self.completed_steps.add(step_name)
def can_run(self, step_name: str) -> tuple:
required = self.STEP_DEPENDENCIES.get(step_name, [])
missing = [r for r in required if r not in self.completed_steps]
if missing:
return False, f'Logic error: {step_name} requires {missing} to complete first'
return True, None
def assert_can_run(self, step_name: str):
ok, error = self.can_run(step_name)
if not ok:
logger.error(error)
raise RuntimeError(error)
guard = AgentStepGuard()
# Simulate trying to skip a step
try:
guard.assert_can_run('create_trello_card')
except RuntimeError as e:
print('Caught logic error:', e)
# Correct sequence
guard.mark_complete('read_email')
guard.mark_complete('extract_action_items')
guard.assert_can_run('create_trello_card') # Now allowed
print('Step sequence valid')结构化错误日志记录
记录每个错误，并提供足够的上下文用于事后复盘分析：错误类型、完整堆栈跟踪、步骤输入，以及失败时任何相关的代理状态。
import logging
import traceback
import json
from datetime import datetime
logger = logging.getLogger('agent.errors')
def log_agent_error(error_type: str, step: str, inputs: dict, exception: Exception, agent_state: dict = None):
error_record = {
'timestamp': datetime.utcnow().isoformat(),
'error_type': error_type,
'step': step,
'exception_type': type(exception).__name__,
'exception_message': str(exception),
'traceback': traceback.format_exc(),
'inputs': inputs,
'agent_state': agent_state or {}
}
logger.error(json.dumps(error_record))
return error_record
# Example usage
try:
raise ValueError('Email ID not found in database')
except Exception as e:
record = log_agent_error(
error_type='data_error',
step='read_email',
inputs={'email_id': 'missing-id-123'},
exception=e,
agent_state={'user_id': 42, 'session_id': 'sess-abc'}
)
print('Error logged:', record['error_type'], '-', record['exception_message'])错误率监控
按步骤和错误类型跟踪错误率，以识别系统性问题。模型错误突然增加，可能表示提示词更改破坏了工具调用；工具错误增加，则可能表示某个 API 性能下降。
from collections import defaultdict
from datetime import datetime
class ErrorTracker:
def __init__(self):
self.errors = defaultdict(list)
def record(self, error_type: str, step: str):
key = f'{error_type}:{step}'
self.errors[key].append(datetime.utcnow())
def get_rates(self, window_minutes: int = 60) -> dict:
from datetime import timedelta
cutoff = datetime.utcnow() - timedelta(minutes=window_minutes)
rates = {}
for key, timestamps in self.errors.items():
recent = [ts for ts in timestamps if ts >= cutoff]
rates[key] = len(recent)
return dict(sorted(rates.items(), key=lambda x: x[1], reverse=True))
def has_spike(self, error_type: str, step: str, threshold: int = 5, window_minutes: int = 10) -> bool:
key = f'{error_type}:{step}'
rates = self.get_rates(window_minutes)
return rates.get(key, 0) >= threshold
tracker = ErrorTracker()
for _ in range(8):
tracker.record('tool_error', 'web_search')
tracker.record('model_error', 'intent_detection')
print('Error rates:', tracker.get_rates())
print('Spike detected:', tracker.has_spike('tool_error', 'web_search'))使用重放调试模型错误
发生模型错误时，使用相同输入重放完全相同的 LLM 调用。将输出与预期结果进行比较。添加更具体的系统提示词说明，或改进工具说明，以修复问题。
import openai
import json
client = openai.OpenAI(api_key='sk-...')
def save_failed_call(step_name: str, messages: list, tools: list, actual_response: str, expected_tool: str, filepath: str):
record = {
'step': step_name,
'messages': messages,
'tools': tools,
'actual_response': actual_response,
'expected_tool': expected_tool
}
with open(filepath, 'w') as f:
json.dump(record, f, indent=2)
print(f'Failed call saved to: {filepath}')
def replay_failed_call(filepath: str, improved_system_prompt: str = None) -> str:
with open(filepath) as f:
record = json.load(f)
messages = record['messages']
if improved_system_prompt:
# Replace system prompt
messages = [m if m['role'] != 'system' else {'role': 'system', 'content': improved_system_prompt}
for m in messages]
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
tools=record['tools']
)
return response.choices[0].message
print('Replay debugging functions defined')根本原因分析清单
代理失败时，请系统地完成以下清单：
- 输入是否有效？（数据错误）
- 是否有外部 API 返回错误？（工具错误）
- LLM 是否调用了正确的工具？参数是否正确？（模型错误）
- 步骤是否按正确顺序执行？（逻辑错误）
- 提示词中是否有足够的上下文？（模型错误——上下文）
def diagnose_failure(error_log: dict) -> dict:
diagnosis = {
'error_type': error_log.get('error_type'),
'root_cause': None,
'immediate_fix': None,
'long_term_fix': None
}
if error_log.get('error_type') == 'data_error':
diagnosis['root_cause'] = 'Invalid or missing input data'
diagnosis['immediate_fix'] = 'Return clear error to caller with field-level validation feedback'
diagnosis['long_term_fix'] = 'Add Pydantic validation at agent entry point'
elif error_log.get('error_type') == 'tool_error':
status = error_log.get('status_code')
if status == 429:
diagnosis['root_cause'] = 'Rate limit hit'
diagnosis['immediate_fix'] = 'Retry with exponential backoff'
diagnosis['long_term_fix'] = 'Add rate limiter to tool calls'
elif status and status >= 500:
diagnosis['root_cause'] = 'Upstream service degradation'
diagnosis['immediate_fix'] = 'Retry up to 3 times, then graceful degradation'
diagnosis['long_term_fix'] = 'Add circuit breaker pattern'
elif error_log.get('error_type') == 'model_error':
diagnosis['root_cause'] = 'LLM tool selection failure'
diagnosis['immediate_fix'] = 'Add explicit tool selection validation'
diagnosis['long_term_fix'] = 'Improve tool descriptions; add few-shot examples'
return diagnosis
result = diagnose_failure({'error_type': 'tool_error', 'status_code': 429})
print('Diagnosis:', result)对严重故障发出警报
并非所有错误都需要立即处理。请按严重程度对错误分类，并相应地分发警报。非关键流程中的静默数据错误可以记录下来；核心流程中断则需要立即发出警报。
ERROR_SEVERITY = {
'data_error': 'low', # Bad input: log and return error to caller
'model_error': 'medium', # LLM misbehavior: investigate, may need prompt fix
'tool_error': 'medium', # API failure: may self-recover with retry
'logic_error': 'high' # Sequencing bug: needs code fix immediately
}
def route_alert(error_type: str, step: str, message: str, is_core_flow: bool = False):
severity = ERROR_SEVERITY.get(error_type, 'medium')
if is_core_flow:
severity = 'high'
if severity == 'high':
print(f'[PAGERDUTY] CRITICAL: {error_type} in {step}: {message}')
# Call PagerDuty API here
elif severity == 'medium':
print(f'[SLACK] WARNING: {error_type} in {step}: {message}')
# Call Slack API here
else:
print(f'[LOG] INFO: {error_type} in {step}: {message}')
route_alert('logic_error', 'create_trello_card', 'Prerequisites not met', is_core_flow=True)
route_alert('data_error', 'parse_input', 'Missing optional field', is_core_flow=False)事后复盘模板
对于严重的代理故障，请撰写事后复盘。一份好的事后复盘应包括：时间线、根本原因、影响、进展顺利之处、出现问题之处，以及防止再次发生的行动项。
def generate_post_mortem(failure_data: dict) -> str:
return f'''
## Post-Mortem: {failure_data.get("title", "Agent Failure")}
**Date**: {failure_data.get("date")}
**Duration**: {failure_data.get("duration_minutes")} minutes
**Impact**: {failure_data.get("impact")}
### Timeline
{chr(10).join(failure_data.get("timeline", []))}
### Root Cause
{failure_data.get("root_cause")}
### Contributing Factors
{chr(10).join(failure_data.get("contributing_factors", []))}
### Action Items
{chr(10).join([f"- [ ] {item}" for item in failure_data.get("action_items", [])])}
'''.strip()
post_mortem_data = {
'title': 'Email Pipeline Failure - Wrong Trello List',
'date': '2025-01-15',
'duration_minutes': 45,
'impact': '200 emails processed but Trello cards created in wrong list',
'timeline': ['09:00 - Pipeline started', '09:15 - First error logged', '09:45 - Fixed and redeployed'],
'root_cause': 'Logic error: TRELLO_LIST_ID env var defaulted to staging value in production',
'contributing_factors': ['- Missing config validation at startup', '- No alert on wrong list ID'],
'action_items': ['Add config validation on startup', 'Alert if list_id changes unexpectedly']
}
print(generate_post_mortem(post_mortem_data))知识检查：根因分析
请测试您对智能体故障根因分析的理解。
调试总结
针对智能体故障进行系统性的根因分析时，可以使用四类分类法：模型错误（LLM 决策问题）、工具错误（接口调用失败）、数据错误（输入数据不正确）和逻辑错误（步骤顺序问题）。此外，还应结合结构化日志记录、错误率监控、针对模型错误的回放调试，以及针对重大故障的事后复盘。
常见问题解答
「分析智能体故障的根本原因」课时是免费的吗？
是的 — 「分析智能体故障的根本原因」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。
「分析智能体故障的根本原因」这节课中我会学到什么？
系统化故障分类：模型错误、工具错误、数据错误、逻辑错误 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Agents 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。
「分析智能体故障的根本原因」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Agents 课中编写并运行代码吗？
能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。