背景/痛点

在 openclaw 项目跑到中后期时，日志问题往往不是“有没有”，而是“能不能用”。我见过不少团队一开始只在控制台打印 print()，本地调试很舒服，等到 Agent 编排、工具调用、模型请求、任务队列、插件扩展都跑起来后，问题就暴露了：

openclaw 的高级玩法通常会涉及多 Agent 协作、工具链调用、异步任务和外部 API 集成。这个阶段日志系统不能只满足“开发看一眼”，而要服务于线上稳定性、成本分析和业务复盘。

我的实践原则是：日志必须结构化、链路必须可追踪、异常必须可聚合、敏感信息必须脱敏。否则日志越多，系统越不可控。

核心内容讲解

构建 openclaw 日志系统，我一般分三层：

1. 应用层日志规范

   • 统一日志格式
   • 统一 trace_id
   • 记录 agent、tool、model、latency 等关键字段

2. 采集与存储

   • 本地写 JSON Lines 文件
   • Filebeat/Fluent Bit 采集
   • 输出到 Elasticsearch、Loki 或 ClickHouse

3. 分析与告警

   • 按 trace_id 追踪任务
   • 按 tool_name 统计失败率
   • 按 model_name 分析调用耗时和成本
   • 对高频异常进行聚合告警

很多人容易忽略一点：openclaw 中最有价值的日志，不是简单的“请求成功/失败”，而是每一步推理与工具执行的过程数据。例如：

{
“trace_id”: “task_20250101_xxx”,
“agent”: “research_agent”,
“tool”: “web_search”,
“event”: “tool_call_finished”,
“latency_ms”: 832,
“success”: true
}

这样的日志可以回答三个问题：任务在哪里慢了？哪个工具最容易失败？哪个 Agent 的调用成本最高？

实战代码/案例

下面以 Python 版本 openclaw 扩展为例，构建一个可落地的结构化日志方案。核心思路是封装一个统一 Logger，所有 Agent、Tool、模型调用都通过它输出日志。

1. 定义结构化日志工具

import json
import time
import uuid
import logging
from contextvars import ContextVar
from datetime import datetime

# 使用 ContextVar 保证异步任务中 trace_id 不串线
trace_id_var = ContextVar("trace_id", default=None)

class OpenClawJsonFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": record.levelname,
            "message": record.getMessage(),
            "logger": record.name,
            "trace_id": trace_id_var.get(),
        }

        # 将 extra 中的业务字段写入 JSON
        if hasattr(record, "extra_fields"):
            log_data.update(record.extra_fields)

        # 异常堆栈单独记录，方便搜索和聚合
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)

        return json.dumps(log_data, ensure_ascii=False)

def init_logger(log_file="openclaw-app.log"):
    logger = logging.getLogger("openclaw")
    logger.setLevel(logging.INFO)

    handler = logging.FileHandler(log_file, encoding="utf-8")
    handler.setFormatter(OpenClawJsonFormatter())

    logger.handlers.clear()
    logger.addHandler(handler)
    return logger

logger = init_logger()

def new_trace_id(prefix="task"):
    trace_id = f"{prefix}_{uuid.uuid4().hex[:12]}"
    trace_id_var.set(trace_id)
    return trace_id

def log_event(level, message, **kwargs):
    """
    统一日志入口，避免业务代码直接拼字符串
    """
    extra = {"extra_fields": kwargs}
    getattr(logger, level)(message, extra=extra)

这里有两个关键点。

第一，`trace_id` 不通过函数参数层层传递，而是使用 `ContextVar`，这对异步 Agent 调度非常友好。

第二，日志输出为 JSON，每一行是一条完整事件，后续无论进 Elasticsearch 还是 Loki，都能直接解析字段。

### 2. 包装 Tool 调用日志

openclaw 中工具调用是最容易出问题的地方，比如搜索超时、数据库连接失败、第三方 API 限流。我们可以封装一个装饰器，对所有工具统一打点。

```python
import functools

def log_tool_call(tool_name):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            start = time.time()

            log_event(
                "info",
                "tool_call_started",
                event="tool_call_started",
                tool=tool_name
            )

            try:
                result = await func(*args, **kwargs)
                latency_ms = int((time.time() - start) * 1000)

                log_event(
                    "info",
                    "tool_call_finished",
                    event="tool_call_finished",
                    tool=tool_name,
                    latency_ms=latency_ms,
                    success=True
                )
                return result

            except Exception as e:
                latency_ms = int((time.time() - start) * 1000)

                log_event(
                    "error",
                    "tool_call_failed",
                    event="tool_call_failed",
                    tool=tool_name,
                    latency_ms=latency_ms,
                    success=False,
                    error_type=type(e).__name__
                )
                raise

        return wrapper
    return decorator

示例工具：

```python
@log_tool_call("customer_profile_query")
async def query_customer_profile(customer_id: str):
    """
    模拟 openclaw 工具：查询客户画像
    """
    await fake_io_wait()

    if customer_id == "bad_id":
        raise ValueError("invalid customer id")

    return {
        "customer_id": customer_id,
        "level": "vip",
        "risk_score": 0.18
    }

async def fake_io_wait():
    import asyncio
    await asyncio.sleep(0.2)

这样做的好处是，业务代码不需要到处写日志，工具层自动输出开始、结束、失败事件。后续分析失败率时，只需要按 `tool` 和 `event` 聚合即可。

### 3. 记录 Agent 执行链路

Agent 任务通常不是一步完成的，而是经过计划、工具调用、模型推理、结果整合。这里我们模拟一个 openclaw 任务入口：

```python
async def run_agent_task(agent_name: str, user_input: str):
    trace_id = new_trace_id("openclaw")

    log_event(
        "info",
        "agent_task_started",
        event="agent_task_started",
        agent=agent_name,
        input_len=len(user_input)
    )

    start = time.time()

    try:
        profile = await query_customer_profile("u_10001")

        # 模拟模型调用日志
        log_event(
            "info",
            "llm_call_finished",
            event="llm_call_finished",
            agent=agent_name,
            model="gpt-4.1",
            prompt_tokens=680,
            completion_tokens=210,
            latency_ms=1260
        )

        latency_ms = int((time.time() - start) * 1000)

        log_event(
            "info",
            "agent_task_finished",
            event="agent_task_finished",
            agent=agent_name,
            latency_ms=latency_ms,
            success=True
        )

        return {
            "trace_id": trace_id,
            "profile": profile,
            "answer": "客户为 VIP，可进入高优先级服务流程"
        }

    except Exception as e:
        latency_ms = int((time.time() - start) * 1000)

        log_event(
            "error",
            "agent_task_failed",
            event="agent_task_failed",
            agent=agent_name,
            latency_ms=latency_ms,
            success=False,
            error_type=type(e).__name__
        )
        raise

一次任务下来，日志会形成这样的链路：

{"event":"agent_task_started","trace_id":"openclaw_xxx","agent":"sales_agent"}
{"event":"tool_call_started","trace_id":"openclaw_xxx","tool":"customer_profile_query"}
{"event":"tool_call_finished","trace_id":"openclaw_xxx","tool":"customer_profile_query","latency_ms":201}
{"event":"llm_call_finished","trace_id":"openclaw_xxx","model":"gpt-4.1","prompt_tokens":680}
{"event":"agent_task_finished","trace_id":"openclaw_xxx","latency_ms":1480}

这就是可分析日志和普通日志的区别。普通日志只能告诉你“执行了”，结构化链路日志能告诉你“每一步怎么执行的”。

### 4. 使用 Fluent Bit 采集日志

生产环境不建议应用直接写远端存储。更稳妥的方式是应用写本地文件，采集组件负责转发。下面是一个简化版 Fluent Bit 配置：

```ini
[SERVICE]
    Flush        5
    Log_Level    info

[INPUT]
    Name         tail
    Path         /var/log/openclaw/openclaw-app.log
    Parser       json
    Tag          openclaw.app
    Refresh_Interval 5

[FILTER]
    Name         modify
    Match        openclaw.app
    Add          service openclaw
    Add          env prod

[OUTPUT]
    Name         es
    Match        openclaw.app
    Host         elasticsearch
    Port         9200
    Index        openclaw-logs

如果团队使用 Loki，也可以把 `OUTPUT` 改成 Loki。选择 Elasticsearch 还是 Loki，我的经验是：如果需要复杂字段检索和聚合分析，Elasticsearch 更顺手；如果主要围绕 Kubernetes 和 Grafana 看日志，Loki 成本更低。

### 5. 常用分析语句

以 Elasticsearch DSL 为例，查询某个任务完整链路：

{
  "query": {
    "term": {
      "trace_id.keyword": "openclaw_abc123"
    }
  },
  "sort": [
    {
      "timestamp": "asc"
    }
  ]
}

统计工具失败率：

{
  "size": 0,
  "query": {
    "term": {
      "event.keyword": "tool_call_failed"
    }
  },
  "aggs": {
    "by_tool": {
      "terms": {
        "field": "tool.keyword",
        "size": 20
      }
    }
  }
}

统计模型调用耗时：

{
  "size": 0,
  "query": {
    "term": {
      "event.keyword": "llm_call_finished"
    }
  },
  "aggs": {
    "by_model": {
      "terms": {
        "field": "model.keyword"
      },
      "aggs": {
        "avg_latency": {
          "avg": {
            "field": "latency_ms"
          }
        }
      }
    }
  }
}

这些查询可以直接沉淀成 Grafana 面板，例如 Agent 成功率、工具失败 Top10、模型平均耗时、Token 消耗趋势。对于 openclaw 这类智能应用框架，这些指标不仅是技术指标，也会直接影响商业成本。

## 总结与思考

openclaw 日志管理的核心，不是把日志“打得更多”，而是让日志具备分析价值。我的建议是从一开始就定义事件模型，比如 `agent_task_started`、`tool_call_finished`、`llm_call_finished`，不要等系统复杂后再补。

落地时有几个经验值得注意：

| 经验 | 原因 |
|---|---|
| 不要记录完整 prompt | 容易泄露隐私，也会撑爆存储 |
| 记录 token 数和耗时 | 便于成本分析和性能优化 |
| 所有任务必须有 trace_id | 没有链路标识，日志价值减半 |
| error 日志必须带 error_type | 方便聚合相同异常 |
| 日志字段要稳定 | 字段频繁变化会破坏分析面板 |

从职业成长角度看，很多程序员只关注功能开发，却忽略可观测性建设。但在真实商业系统里，能不能快速定位线上问题，往往决定了你是否具备负责复杂系统的能力。openclaw 这类 Agent 系统更是如此，模型的不确定性、工具的不稳定性、外部 API 的波动都会放大排障难度。

一个成熟的日志系统，不只是运维工具，而是研发团队理解系统行为的基础设施。它能帮助我们发现高成本调用、识别低质量工具、优化 Agent 编排策略，也能在故障发生时把排查时间从几个小时压缩到几分钟。对 openclaw 的高级实践来说，这一步投入非常值得。




#云盏科技官网 #小龙虾 #云盏科技 #ai技术论坛 #skills市场