主题
原文：https://www.mstx.cn/harness/417.html · 创建：2026-09-03 · 更新：2026-09-05 · 阅读：1082
一个熟悉的场景
你在开发Agent，一切都看起来很顺利。然后某一天，用户报告了一个问题："Agent给出了一个奇怪的回答。"
你问："什么回答？"
用户说："我不记得了，大概是说了一些不相关的东西。"
你打开终端，想查看日志。但你发现——你的Agent没有日志。或者说，只有几条print语句的输出，而且这些输出在终端关闭后就消失了。
你不知道：
- 用户当时输入了什么
- Agent经历了多少轮循环
- 每一轮LLM返回了什么
- 工具调用了什么参数、返回了什么结果
- 最终的回答是怎么得出的
你无法复现问题，因此无法修复问题。
这就是本节要解决的核心问题：如何让Agent的执行过程可见、可追踪、可诊断。
日志的三个层次
Agent的日志体系应该覆盖三个层次：
第一层：系统级日志——记录Harness的运行状态（启动、关闭、配置加载、错误）。
第二层：执行级日志——记录每次Agent执行的完整轨迹（输入、循环过程、工具调用、输出）。
第三层：调试级日志——记录足够详细的内部信息，用于定位复杂问题（完整消息内容、token消耗、每次API调用的原始响应）。
这三个层次各有用途。系统级日志让你知道"服务是否正常"，执行级日志让你知道"Agent做了什么"，调试级日志让你知道"Agent为什么这样做"。
使用Python的logging模块
告别print，使用Python标准库的logging：
python
# harness/logger.py
import logging
import sys
from typing import Optional
def setup_logger(name: str = "harness",
level: str = "INFO",
log_file: Optional[str] = None) -> logging.Logger:
"""创建并配置一个logger"""
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper()))
# 清除已有的handler（避免重复）
logger.handlers.clear()
# 控制台handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(getattr(logging, level.upper()))
console_format = logging.Formatter(
'%(asctime)s | %(levelname)-7s | %(name)s | %(message)s',
datefmt='%H:%M:%S'
)
console_handler.setFormatter(console_format)
logger.addHandler(console_handler)
# 文件handler（如果指定了日志文件）
if log_file:
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_handler.setLevel(logging.DEBUG) # 文件记录所有级别
file_format = logging.Formatter(
'%(asctime)s | %(levelname)-7s | %(name)s | %(filename)s:%(lineno)d | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
file_handler.setFormatter(file_format)
logger.addHandler(file_handler)
return logger这个setup_logger函数提供了两个输出通道：
- 控制台：输出指定级别以上的日志（默认INFO），用于开发时观察
- 文件：输出所有级别的日志（DEBUG及以上），用于事后排查
关键设计：文件日志的格式比控制台更详细（包含文件名和行号），因为文件日志用于事后排查，需要更多上下文信息。
执行轨迹的专门日志
除了通用的logging，Agent还需要一个专门的轨迹记录器。它记录的不是"系统发生了什么"，而是"Agent的这次执行经历了什么"。
python
# harness/tracer.py
import json
import time
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field, asdict
@dataclass
class StepRecord:
"""一次循环迭代的记录"""
iteration: int
step_type: str # "llm_call" | "tool_execution" | "final_answer"
timestamp: float
details: Dict[str, Any] = field(default_factory=dict)
@dataclass
class ExecutionTrace:
"""一次完整Agent执行的轨迹"""
trace_id: str
start_time: float
end_time: Optional[float] = None
user_input: str = ""
steps: List[StepRecord] = field(default_factory=list)
final_status: str = ""
termination_reason: str = ""
def add_step(self, step: StepRecord):
self.steps.append(step)
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
def to_json(self) -> str:
return json.dumps(self.to_dict(), ensure_ascii=False, indent=2)
class Tracer:
"""执行轨迹记录器"""
def __init__(self, logger=None):
self.logger = logger
self.current_trace: Optional[ExecutionTrace] = None
self.traces: List[ExecutionTrace] = []
def start_trace(self, user_input: str) -> ExecutionTrace:
"""开始记录一次新的执行"""
import uuid
trace = ExecutionTrace(
trace_id=str(uuid.uuid4())[:8],
start_time=time.time(),
user_input=user_input,
)
self.current_trace = trace
if self.logger:
self.logger.info(f"[Trace {trace.trace_id}] 开始执行，用户输入：{user_input[:100]}")
return trace
def record_llm_call(self, iteration: int,
input_messages_count: int,
output_type: str,
output_preview: str,
duration_ms: int):
"""记录一次LLM调用"""
step = StepRecord(
iteration=iteration,
step_type="llm_call",
timestamp=time.time(),
details={
"input_messages_count": input_messages_count,
"output_type": output_type, # "tool_calls" | "final_answer"
"output_preview": output_preview[:200],
"duration_ms": duration_ms,
}
)
if self.current_trace:
self.current_trace.add_step(step)
if self.logger:
self.logger.debug(f"[迭代{iteration}] LLM调用：{output_type}，耗时{duration_ms}ms")
def record_tool_call(self, iteration: int,
tool_name: str,
arguments: Dict[str, Any],
result_preview: str,
success: bool,
duration_ms: int):
"""记录一次工具调用"""
step = StepRecord(
iteration=iteration,
step_type="tool_execution",
timestamp=time.time(),
details={
"tool_name": tool_name,
"arguments": arguments,
"result_preview": result_preview[:200],
"success": success,
"duration_ms": duration_ms,
}
)
if self.current_trace:
self.current_trace.add_step(step)
if self.logger:
status = "成功" if success else "失败"
self.logger.info(
f"[迭代{iteration}] 工具 {tool_name} {status}，"
f"参数：{json.dumps(arguments, ensure_ascii=False)[:100]}，"
f"耗时{duration_ms}ms"
)
def end_trace(self, status: str, termination_reason: str):
"""结束当前执行轨迹"""
if self.current_trace:
self.current_trace.end_time = time.time()
self.current_trace.final_status = status
self.current_trace.termination_reason = termination_reason
self.traces.append(self.current_trace)
if self.logger:
duration = self.current_trace.end_time - self.current_trace.start_time
self.logger.info(
f"[Trace {self.current_trace.trace_id}] 执行结束："
f"状态={status}，迭代={len(self.current_trace.steps)}，"
f"耗时={duration:.2f}秒，终止原因={termination_reason}"
)
self.current_trace = None
def save_trace(self, filepath: str):
"""将当前轨迹保存到文件"""
if self.current_trace:
with open(filepath, 'w', encoding='utf-8') as f:
f.write(self.current_trace.to_json())这个Tracer的设计有几个关键点：
第一：轨迹是结构化的。 每一步的记录都有明确的字段（迭代次数、步骤类型、时间戳、详细信息）。这意味着轨迹数据可以被程序化地分析，而不仅仅是被人眼阅读。
第二：轨迹与日志是分离的。 Tracer记录"Agent做了什么"，logging记录"系统发生了什么"。两者有不同的用途和格式。轨迹是结构化的数据，日志是人类可读的文本。
第三：轨迹有唯一ID。 每次执行都有一个trace_id，这让"用户报告的问题"能够与"对应的执行轨迹"精确对应。
将Tracer集成到Runtime
修改AgentRuntime，在循环的关键位置添加轨迹记录：
python
class AgentRuntime:
def __init__(self, config, tool_registry, memory,
output_schema=None, tracer=None, logger=None):
self.config = config
self.tools = tool_registry
self.memory = memory
self.output_schema = output_schema
self.tracer = tracer
self.logger = logger or logging.getLogger("harness")
self.client = OpenAI(
base_url=config.llm.base_url,
api_key=config.llm.api_key,
)
def run(self, user_input: str) -> AgentResult:
start_time = time.time()
# 开始轨迹
trace = None
if self.tracer:
trace = self.tracer.start_trace(user_input)
self.memory.initialize(user_input)
for iteration in range(self.config.runtime.max_iterations):
elapsed = time.time() - start_time
if elapsed > self.config.runtime.timeout_seconds:
result = AgentResult(
status="timeout",
iterations=iteration,
duration_ms=int(elapsed * 1000),
termination_reason=f"超时（>{self.config.runtime.timeout_seconds}秒）"
)
if self.tracer:
self.tracer.end_trace(result.status, result.termination_reason)
return result
# LLM调用
llm_start = time.time()
try:
api_params = {
"model": self.config.llm.model,
"messages": self.memory.get_context(),
"temperature": self.config.llm.temperature,
"max_tokens": self.config.llm.max_tokens,
}
tool_schemas = self.tools.get_schemas()
if tool_schemas:
api_params["tools"] = tool_schemas
if self.output_schema:
api_params["response_format"] = {
"type": "json_schema",
"json_schema": {
"name": "agent_output",
"schema": self.output_schema,
"strict": True,
}
}
response = self.client.chat.completions.create(**api_params)
except Exception as e:
elapsed = time.time() - start_time
result = AgentResult(
status="error",
content=f"LLM调用失败：{e}",
iterations=iteration,
duration_ms=int(elapsed * 1000),
termination_reason="LLM调用异常"
)
if self.tracer:
self.tracer.end_trace(result.status, result.termination_reason)
return result
llm_duration = int((time.time() - llm_start) * 1000)
message = response.choices[0].message
# 记录LLM调用
if self.tracer:
if message.tool_calls:
output_type = "tool_calls"
preview = f"{len(message.tool_calls)}个工具调用"
else:
output_type = "final_answer"
preview = (message.content or "")[:100]
self.tracer.record_llm_call(
iteration=iteration + 1,
input_messages_count=len(self.memory.get_context()),
output_type=output_type,
output_preview=preview,
duration_ms=llm_duration,
)
if message.tool_calls:
self.memory.add_llm_message(message)
calls = self._parse_tool_calls(message)
for call in calls:
# 执行工具
tool_start = time.time()
result = self.tools.execute(call.name, call.arguments)
tool_duration = int((time.time() - tool_start) * 1000)
# 记录工具调用
if self.tracer:
self.tracer.record_tool_call(
iteration=iteration + 1,
tool_name=call.name,
arguments=call.arguments,
result_preview=result.content[:100] if result.success else result.error,
success=result.success,
duration_ms=tool_duration,
)
# 构建反馈
feedback = self._build_tool_feedback(call, result)
self.memory.messages.append(feedback)
else:
# 正常终止
elapsed = time.time() - start_time
content = message.content or ""
parsed_output = None
if self.output_schema and content:
try:
parsed_output = json.loads(content)
except json.JSONDecodeError:
pass
result = AgentResult(
status="success",
content=content,
structured_output=parsed_output,
iterations=iteration + 1,
duration_ms=int(elapsed * 1000),
termination_reason="正常终止"
)
if self.tracer:
self.tracer.end_trace(result.status, result.termination_reason)
return result
# 达到最大迭代次数
elapsed = time.time() - start_time
result = AgentResult(
status="max_iterations",
content="已达到最大循环次数",
iterations=self.config.runtime.max_iterations,
duration_ms=int(elapsed * 1000),
termination_reason=f"达到最大迭代次数（{self.config.runtime.max_iterations}次）"
)
if self.tracer:
self.tracer.end_trace(result.status, result.termination_reason)
return result调试的实用技巧
有了日志和轨迹，调试Agent就变得可行了。以下是几个实用的调试技巧：
技巧一：从轨迹中定位问题迭代
当Agent给出错误回答时，不要从最终答案开始分析。从轨迹中找出"哪一步开始偏了"。
python
def analyze_trace(trace: ExecutionTrace):
"""简单分析一次执行轨迹，找出异常步骤"""
print(f"\n=== 轨迹分析：{trace.trace_id} ===")
print(f"用户输入：{trace.user_input}")
print(f"最终状态：{trace.final_status}")
print(f"终止原因：{trace.termination_reason}")
print()
for step in trace.steps:
if step.step_type == "llm_call":
print(f" 迭代{step.iteration} | LLM调用 | {step.details['output_type']} | {step.details['duration_ms']}ms")
print(f" 输出预览：{step.details['output_preview']}")
elif step.step_type == "tool_execution":
status = "✓" if step.details['success'] else "✗"
print(f" 迭代{step.iteration} | 工具 {step.details['tool_name']} {status} | {step.details['duration_ms']}ms")
print(f" 参数：{json.dumps(step.details['arguments'], ensure_ascii=False)[:100]}")
print(f" 结果：{step.details['result_preview'][:100]}")
print()技巧二：失败工具调用的重点检查
当Agent失败时，优先检查执行失败的工具调用。一个工具失败可能导致后续所有推理都建立在错误的信息上。
技巧三：关注循环次数的异常
- 循环次数为1：Agent没有调用任何工具就给出了回答。可能的原因：工具描述不清晰，LLM不知道可以调用工具；或者任务确实不需要工具。
- 循环次数等于最大迭代次数：Agent陷入了循环。可能的原因：工具返回的信息不足以让LLM推进任务；或者任务本身对当前工具集来说不可完成。
技巧四：保存"坏案例"的完整轨迹
每次发现Agent表现不佳时，保存完整的轨迹和用户输入。这些"坏案例"是改进Agent的最宝贵资源——它们直接告诉你Agent在哪里失败、为什么失败。第7章的评估体系将建立在这个积累之上。
日志级别的使用指南
| 级别 | 用途 | 示例 |
|---|---|---|
| DEBUG | 详细的内部信息，用于定位复杂问题 | 完整的消息内容、API原始响应 |
| INFO | 关键的执行信息，用于了解Agent做了什么 | 工具调用、循环迭代、执行结果 |
| WARNING | 潜在问题，不影响执行但值得注意 | Schema描述过短、工具结果被截断 |
| ERROR | 执行错误，需要关注 | LLM调用失败、工具执行异常 |
| CRITICAL | 系统级故障，需要立即处理 | 配置加载失败、无法连接Provider |
经验法则： 开发环境用DEBUG级别（看到所有细节），生产环境用INFO级别（看到关键事件，避免日志爆炸）。
日志的性能考量
日志本身是有成本的。过多的日志会：
- 消耗磁盘空间
- 降低执行速度
- 让真正的错误信息淹没在噪音中
建议：
- 生产环境中，INFO级别的日志只记录关键事件（执行开始/结束、工具调用、错误）
- DEBUG级别的完整消息内容只在需要时开启
- 轨迹数据的保存异步进行，不阻塞主流程
本节小结
日志与调试是Harness可观测性的基础。本节实现了：
- setup_logger：配置多通道的日志输出（控制台+文件）
- Tracer：记录结构化的执行轨迹，包括每一步的详细信息
- 轨迹分析工具：从轨迹中定位问题
核心原则是：日志让你知道发生了什么，轨迹让你理解为什么发生。 两者配合，Agent的执行过程就不再是一个黑盒。
在下一节中，我们将进入第4章，深入讨论工具系统设计——如何构建安全、健壮、可维护的工具生态。