读完你能直接把“turn_id / tokens / tool / latency”这些关键字段写进 JSON 日志，并用一段 Python 在 10 秒内定位最费 token 的轮次。

你可能遇到过：Agent 一开始很稳，过一阵子开始不稳定；你去查原因，日志只有 Turn 1 ok 这种信息。

问题不在“日志写得少”，而在日志没有结构。你没法查询、没法聚合、没法做回归对比。

本文给 5 个工程技巧，全部配可运行代码，并带一个真实本地小实验（结果文件在 experiment/results.json / experiment/results_advanced.json）。

---

0. 什么叫“可调试”的 Agent 日志

至少满足三点：

1. 每轮都有稳定字段（turn、tokens、tool、latency_ms）
2. 输出可查询（能快速找异常轮次/失败率）
3. 输出可治理（长字段截断、敏感字段脱敏）

---

1) 技巧 1：固定字段先立起来

推荐最小字段：trace_id / turn / query / tokens / tool / tool_success / latency_ms / level。

# agent_loop_minimal.py
import json
import random
import time

TOOLS = ["search_api", "weather_api", "calculator", None]

def run_agent(n_turns: int = 5, trace_id: str = "t-001"):
    for turn in range(1, n_turns + 1):
        event = {
            "trace_id": trace_id,
            "turn": turn,
            "query": f"用户问题 #{turn}",
            "tokens": random.randint(30, 200),
            "tool": random.choice(TOOLS),
            "tool_success": True,
            "latency_ms": round(random.uniform(120, 800), 1),
            "level": "info",
        }
        print(json.dumps(event, ensure_ascii=False))
        time.sleep(0.02)

if __name__ == "__main__":
    run_agent()

---

2) 技巧 2：structlog processor 链做治理（截断 + 脱敏）

# structlog_agent.py
import re
import structlog


def truncate_long_fields(max_len: int = 200):
    def _p(logger, method_name, event_dict):
        for k, v in list(event_dict.items()):
            if isinstance(v, str) and len(v) > max_len:
                event_dict[k] = v[:max_len] + "...<truncated>"
        return event_dict
    return _p


def mask_secrets():
    secret_re = re.compile(r"(sk-|token=)[A-Za-z0-9_-]+")
    def _p(logger, method_name, event_dict):
        for k, v in list(event_dict.items()):
            if isinstance(v, str):
                event_dict[k] = secret_re.sub(r"\1***", v)
        return event_dict
    return _p


structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.stdlib.add_log_level,
        mask_secrets(),
        truncate_long_fields(240),
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)


def demo():
    log = structlog.get_logger("agent")
    log.info(
        "agent_turn",
        trace_id="t-002",
        turn=1,
        tokens=128,
        tool="search_api",
        tool_success=True,
        latency_ms=312.4,
        prompt="x" * 500,
        note="token=abcdefg123456",
    )

if __name__ == "__main__":
    demo()

---

3) 技巧 3：loguru sink 分流（终端可读 + JSON 可查）

# loguru_agent.py
import json
from loguru import logger


def json_sink(message):
    r = message.record
    e = r["extra"]
    entry = {
        "time": r["time"].isoformat(),
        "level": r["level"].name,
        "trace_id": e.get("trace_id"),
        "turn": e.get("turn"),
        "tokens": e.get("tokens"),
        "tool": e.get("tool"),
        "tool_success": e.get("tool_success"),
        "latency_ms": e.get("latency_ms"),
        "event": r["message"],
    }
    print(json.dumps(entry, ensure_ascii=False))


def demo():
    logger.remove()
    logger.add(lambda m: print(m, end=""), format="{time} {level} {message} {extra}\n")
    logger.add(json_sink, format="{message}")

    logger.bind(trace_id="t-003", turn=1, tokens=88, tool="weather_api", tool_success=True, latency_ms=210.5).info(
        "agent_turn"
    )

if __name__ == "__main__":
    demo()

---

4) 技巧 4：用一段 Python 直接查出“最费 token 的轮次”

# analyze_results.py
import json
from pathlib import Path


def main():
    data = json.loads(Path("experiment/results_advanced.json").read_text(encoding="utf-8"))
    turns = data["turns_log"]

    heavy = sorted([t for t in turns if t["tokens"] > 100], key=lambda x: x["tokens"], reverse=True)
    print("token>100 的轮次：")
    for t in heavy:
        print(f"- turn={t['turn']} tokens={t['tokens']} tool={t['tool']} latency_ms={t['latency_ms']}")

if __name__ == "__main__":
    main()

---

5) 技巧 5：contextvars 跨 async 传播 trace_id

# contextvars_trace.py
import asyncio
import contextvars

trace_id_var = contextvars.ContextVar("trace_id", default="-")

def log(msg: str):
    print(f"trace_id={trace_id_var.get()} {msg}")

async def worker(name: str):
    log(f"start {name}")
    await asyncio.sleep(0.05)
    log(f"end {name}")

async def main():
    token = trace_id_var.set("t-ctx-001")
    try:
        await asyncio.gather(worker("a"), worker("b"))
    finally:
        trace_id_var.reset(token)

if __name__ == "__main__":
    asyncio.run(main())

---

本地小实验：结果文件在哪、怎么看

本文 run 目录里有两份结果文件：

• experiment/results.json：四种方案的 3 轮对比（含耗时、JSON 可解析性）
• experiment/results_advanced.json：10 轮模拟日志 + 统计分析

如果你只想验证“结构化日志能不能被快速查询”，直接跑 analyze_results.py 就能打印 token>100 的轮次。

---

对比表：4 种方案怎么选（含本地实验数据）

方案	JSON 可解析	字段可查询	3 轮开销（ms）	结论
print	否	否	0.021	只适合临时 debug
stdlib logging	否	否	0.111	结构化支持弱
structlog	是	是	0.272	适合做截断/脱敏治理
loguru	是	是	2.518	上手快，文件管理友好（demo 含初始化）

数据来源：experiment/results.json

---

常见坑（你很可能会踩一次）

1. 把 prompt/工具返回原样写进日志：日志会爆炸，查询也会变慢。解决：默认截断（技巧 2）。
2. 字段名不稳定：今天叫 turn_id，明天叫 step，后天叫 round，统计脚本直接失效。解决：固定字段集合。
3. 没有 trace_id：并发一上来，你根本串不起来同一次请求。解决：contextvars（技巧 5）。

---

FAQ

Q1：结构化日志会不会太慢？

慢的不是 JSON，慢的是你把超长 prompt/工具返回原样写进去了。建议默认截断 + 脱敏。

Q2：tokens 字段怎么拿？

先把字段结构固定住：没有真实 usage 时可以先用估算占位；等接入真实返回后再替换。

Q3：能不能记录 prompt？

可以，但只记录必要片段，并且默认截断。

---

你今天就能落地的最小版本

• 固定字段：turn/tokens/tool/latency_ms
• 输出 JSON 可解析

做到这两点，Agent 调试会从“翻一堆文本”变成“查一条事件流”。