Agent调用工具时如何优雅输出描述：结构化日志与上下文管理实践

1次阅读

共计 4405 个字符，预计需要花费 12 分钟才能阅读完成。

最近在开发一个多 Agent 协作系统时，遇到一个典型问题：当多个工具被并行调用时，控制台输出的日志像一锅乱炖。比如下面这个电商订单处理场景：

# 伪代码示例
def process_order(order):
    payment_result = payment_gateway.charge(order)  # 输出："处理支付中..."
    inventory_result = warehouse.check_stock(order)  # 输出："库存检查开始"
    logistics_result = shipper.book_delivery(order)  # 输出："[ERROR] 地址校验失败"

遇到问题时，我们不得不像侦探一样：

无法快速定位某个错误属于哪个订单的哪个处理环节
不同工具的输出格式五花八门（有的带时间戳，有的只有纯文本）
敏感数据（如用户手机号）直接暴露在日志中

# 最朴素的实现方式
print(f"[{datetime.now()}] 调用支付网关，订单: {order.id}")

优点：零学习成本
缺点：
格式不统一难以解析
上下文信息需要手动拼接
改日志格式需要全局搜索替换

@log_tool_call(tool_name="payment")
def charge(order):
    pass

优点：业务代码保持干净
缺点：
每个工具方法需要单独装饰
难以获取完整的调用链信息

通过动态代理在工具调用前后插入日志逻辑：

# 伪代码展示核心思想
class ToolProxy:
    def __call__(self, tool_method, *args):
        start_ctx = build_context(args)
        logger.info(f"tool_call_start", extra=start_ctx)

        try:
            result = tool_method(*args)
            end_ctx = build_context(result)
            logger.info(f"tool_call_end", extra=end_ctx)
            return result
        except Exception as e:
            logger.error(f"tool_call_fail", extra={"error": str(e)})
            raise

定义所有工具必须遵守的日志 Schema：

// 日志 JSON Schema
{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "required": ["trace_id", "span_id", "timestamp"],
  "properties": {"trace_id": {"type": "string"},  // 全局唯一调用链 ID
    "span_id": {"type": "string"},   // 当前工具调用 ID
    "parent_id": {"type": "string"}, // 父调用 ID
    "timestamp": {"type": "string"},
    "tool_name": {"type": "string"},
    "input": {"type": "object"},    // 输入参数（自动脱敏）"output": {"type": "object"},   // 输出结果
    "duration_ms": {"type": "number"}
  }
}

完整的生产级工具调用封装：

import contextvars
import json
import time
from functools import wraps
from typing import Any, Dict

# 上下文变量
current_trace = contextvars.ContextVar("trace_id", default="")
current_span = contextvars.ContextVar("span_id", default="")

class ToolLogger:
    @classmethod
    def log_call(cls, tool_name: str):
        """工具调用日志装饰器"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # 生成调用链 ID
                trace_id = current_trace.get() or generate_id()
                parent_id = current_span.get()
                span_id = generate_id()

                # 设置上下文
                token = current_span.set(span_id)
                start_time = time.time()

                # 构造日志上下文
                log_ctx = {
                    "trace_id": trace_id,
                    "span_id": span_id,
                    "parent_id": parent_id,
                    "tool_name": tool_name,
                    "timestamp": get_iso_time(),
                    "input": cls._sanitize_input(kwargs)
                }

                logger.info("tool_call_start", extra=log_ctx)

                try:
                    result = func(*args, **kwargs)
                    log_ctx.update({"output": cls._sanitize_output(result),
                        "duration_ms": (time.time() - start_time) * 1000
                    })
                    logger.info("tool_call_success", extra=log_ctx)
                    return result
                except Exception as e:
                    log_ctx["error"] = str(e)
                    logger.error("tool_call_failed", extra=log_ctx)
                    raise
                finally:
                    current_span.reset(token)

            return wrapper
        return decorator

    @staticmethod
    def _sanitize_input(data: Dict) -> Dict:
        """敏感信息脱敏"""
        sanitized = data.copy()
        if "phone" in sanitized:
            sanitized["phone"] = "****" + sanitized["phone"][-4:]
        return sanitized

通过环境变量动态调整日志级别：

import logging
import os

LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()

class DynamicLogger:
    def __init__(self, name):
        self.logger = logging.getLogger(name)
        self.set_level(LOG_LEVEL)

    def set_level(self, level):
        """运行时动态调整日志级别"""
        if level == "DEBUG":
            self.logger.setLevel(logging.DEBUG)
        elif level == "WARN":
            self.logger.setLevel(logging.WARNING)
        else:
            self.logger.setLevel(logging.INFO)

方案	序列化速度	数据体积	可读性
JSON	1x	1x	高
MessagePack	3x	0.7x	低
Protobuf	5x	0.5x	无

建议：

开发环境使用 JSON 便于调试
生产环境使用 MessagePack 平衡性能与可维护性

使用多进程队列避免阻塞主线程：

from concurrent.futures import ThreadPoolExecutor
import queue

log_queue = queue.Queue(maxsize=1000)

def async_log_worker():
    while True:
        try:
            log_entry = log_queue.get()
            write_to_elasticsearch(log_entry)  # 实际写入操作
        except Exception as e:
            print(f"Log worker error: {e}")

# 启动后台线程
with ThreadPoolExecutor(max_workers=2) as executor:
    executor.submit(async_log_worker)

当工具 A 调用工具 B，工具 B 又回调工具 A 时：

# 解决方案：增加调用深度检测
def tool_call(func):
    call_depth = 0
    MAX_DEPTH = 10

    @wraps(func)
    def wrapper(*args):
        nonlocal call_depth
        if call_depth > MAX_DEPTH:
            raise RecursionError("Maximum call depth exceeded")
        call_depth += 1
        try:
            return func(*args)
        finally:
            call_depth -= 1

Python 的 contextvars 在新建线程时会丢失上下文，需要手动传递：

import threading

def run_in_thread(func, *args):
    # 捕获当前上下文
    context = {"trace_id": current_trace.get(),
        "span_id": current_span.get()}

    def wrapped():
        # 还原上下文
        current_trace.set(context["trace_id"])
        current_span.set(context["span_id"])
        return func(*args)

    threading.Thread(target=wrapped).start()

# 按日志级别配置不同存储策略
LOG_CONFIG = {
    "INFO": {
        "storage": "elasticsearch",
        "retention_days": 7
    },
    "ERROR": {"storage": ["elasticsearch", "s3"],
        "retention_days": 30
    }
}

如果无法修改第三方工具的代码，如何实现输出格式统一？这里有几个思路供讨论：