🧠 图解记忆：统一事件字段、Trace 关联、输入输出摘要与错误分类，日志才可搜索、聚合和审计；点击图片可查看原图。
💡 答案要点
日志设计原则：
传统日志：文本格式，难以搜索
结构化日志：JSON 格式，可查询、可分析、可告警
Agent 日志特殊要求：
① TraceID 关联（同一请求的所有日志）
② 步骤追踪（每个 Tool 调用是独立的 Span）
③ 上下文保存（中间状态的 Thought/Action/Observation）
④ 性能埋点（延迟、Token 消耗、成本）结构化日志 Schema：
展开 Python 代码示例（77 行）
python
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, List, Dict, Any
import json
@dataclass
class AgentLogEntry:
"""Agent 结构化日志条目"""
# 基础字段（必须）
timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
level: str = "INFO" # DEBUG/INFO/WARNING/ERROR
trace_id: str = "" # 跨请求唯一
span_id: str = "" # 当前步骤唯一
# Agent 上下文
agent_name: str = ""
agent_version: str = ""
task_id: str = ""
session_id: str = ""
# 步骤信息
step_index: int = 0
step_type: str = "" # "thought"/"action"/"observation"/"result"
# LLM 调用信息
model: str = ""
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
latency_ms: int = 0
cost_usd: float = 0.0
# Tool 调用信息
tool_name: str = ""
tool_args: Dict[str, Any] = field(default_factory=dict)
tool_result: Any = None
tool_error: Optional[str] = None
# 业务信息
user_id: str = ""
intent: str = "" # 用户意图分类
success: bool = True
error_message: Optional[str] = None
# 可扩展字段
extra: Dict[str, Any] = field(default_factory=dict)
def to_json(self) -> str:
"""序列化为 JSON"""
return json.dumps(self.__dict__, ensure_ascii=False, default=str)
@classmethod
def from_json(cls, json_str: str) -> "AgentLogEntry":
"""反序列化"""
return cls(**json.loads(json_str))
def to_otel_span(self) -> dict:
"""转换为 OpenTelemetry Span 格式"""
return {
"trace_id": self.trace_id,
"span_id": self.span_id,
"parent_span_id": self.parent_span_id if hasattr(self, "parent_span_id") else "",
"operation_name": f"{self.agent_name}.{self.step_type}",
"start_time": self.timestamp,
"duration_ms": self.latency_ms,
"tags": {
"agent_name": self.agent_name,
"step_index": self.step_index,
"model": self.model,
"tool_name": self.tool_name,
"success": str(self.success),
},
"logs": [
{"timestamp": self.timestamp, "fields": self.__dict__}
]
}日志采集架构：
展开 Python 代码示例（49 行）
python
import logging
from opentelemetry import trace
from logging.handlers import RotatingFileHandler
import json
class AgentJSONFormatter(logging.Formatter):
"""JSON 格式日志 formatter"""
def format(self, record: logging.LogRecord) -> str:
log_data = {
"timestamp": self.formatTime(record, self.datefmt),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"trace_id": self._get_trace_id(),
"span_id": self._get_span_id(),
"agent_name": getattr(record, "agent_name", ""),
"step_index": getattr(record, "step_index", 0),
"model": getattr(record, "model", ""),
"tool_name": getattr(record, "tool_name", ""),
"success": getattr(record, "success", True),
}
# 添加额外字段
if hasattr(record, "extra"):
log_data.update(record.extra)
return json.dumps(log_data, ensure_ascii=False)
def _get_trace_id(self) -> str:
span = trace.get_current_span()
if span:
ctx = span.get_span_context()
return format(ctx.trace_id, "032x") if ctx else ""
return ""
def _get_span_id(self) -> str:
span = trace.get_current_span()
if span:
ctx = span.get_span_context()
return format(ctx.span_id, "016x") if ctx else ""
return ""
# 配置日志
logger = logging.getLogger("agent")
logger.setLevel(logging.INFO)
handler = RotatingFileHandler("/var/log/agent/app.log", maxBytes=100_000_000, backupCount=10)
handler.setFormatter(AgentJSONFormatter())
logger.addHandler(handler)日志查询示例（Elasticsearch）：
展开 Python 代码示例（74 行）
python
# 查询某个 TraceID 的所有日志
QUERY_TRACE = """
{
"query": {
"bool": {
"must": [
{"match": {"trace_id": "abc123def456"}}
]
}
},
"sort": [{"timestamp": "asc"}],
"size": 1000
}
"""
# 查询所有失败的 Tool 调用
QUERY_FAILED_TOOLS = """
{
"query": {
"bool": {
"must": [
{"match": {"level": "ERROR"}},
{"match": {"tool_name": "search_database"}},
{"range": {"timestamp": {"gte": "now-1h"}}}
]
}
}
}
"""
# 查询 Token 消耗异常（> 10K）
QUERY_HIGH_TOKEN = """
{
"query": {
"bool": {
"must": [
{"range": {"total_tokens": {"gt": 10000}}},
{"range": {"timestamp": {"gte": "now-1d"}}}
]
}
},
"aggs": {
"by_agent": {
"terms": {"field": "agent_name"},
"aggs": {
"avg_tokens": {"avg": {"field": "total_tokens"}},
"max_tokens": {"max": {"field": "total_tokens"}},
"p95_tokens": {"percentiles": {"field": "total_tokens", "percents": [95]}}
}
}
}
}
"""
# 统计每小时的错误率
QUERY_ERROR_RATE = """
{
"query": {
"bool": {
"must": [
{"match": {"level": "ERROR"}},
{"range": {"timestamp": {"gte": "now-24h"}}}
]
}
},
"aggs": {
"hourly_errors": {
"date_histogram": {
"field": "timestamp",
"interval": "hour"
}
}
}
}日志告警规则：
展开 Yaml 代码示例（34 行）
yaml
# Prometheus Alert 规则（基于日志 metrics）
groups:
- name: agent-log-alerts
rules:
# 高错误率告警
- alert: AgentHighErrorRate
expr: |
sum(rate(agent_log_errors_total[5m])) by (agent_name)
/ sum(rate(agent_log_total[5m])) by (agent_name) > 0.05
for: 5m
annotations:
summary: "Agent {{ $labels.agent_name }} 错误率超过 5%"
description: "最近 5 分钟错误率 {{ $value }}"
# Tool 调用超时告警
- alert: AgentToolTimeout
expr: |
histogram_quantile(0.99,
rate(agent_tool_latency_seconds_bucket[5m])
) > 10
for: 3m
annotations:
summary: "Tool 调用 P99 延迟超过 10 秒"
# Token 消耗异常告警
- alert: AgentHighTokenConsumption
expr: |
sum(rate(agent_token_total[1h])) by (agent_name)
> 1.5 * avg_over_time(
sum(rate(agent_token_total[1h])) by (agent_name)[7d:1h]
)
for: 15m
annotations:
summary: "Agent {{ $labels.agent_name }} Token 消耗异常"面试话术：
"我的 Agent 日志设计核心是'结构化 + 可追溯'。每个日志条目包含 trace_id、span_id、step_index、model、tool_name，每次 LLM 调用都记录 token 消耗和延迟。这样做的好处是：① 查问题快——输入 trace_id 就能看到整个请求链路；② 分析容易——用 Elasticsearch 按 agent_name、tool_name、success 聚合；③ 告警准——错误率/延迟超过阈值自动通知。生产环境我每天查日志 < 10 次，但每次都能 5 分钟内定位问题。"