ログと可観測性

アプリを公開すると、ユーザーはあなたの見えないところでそれを使います。ある日誰かが「そちらのアシスタントがまったく間違ったことを答えた」と知らせてきたり、月末の請求が予想の 3 倍だったりしたら、どうやって調べますか。

手元にユーザーの質問と最終的な回答しかなければ、ほとんど手のつけようがありません。検索が正しいドキュメントを見つけられなかったのか。モデルが読み違えたのか。あるツールがエラーを出したのか。それとも、あるステップで道を外れ、その後どんどん間違いが大きくなったのか。

可観測性（observability）とは、システムが動いているあいだに十分な記録を残し、後から実際に何をしたのかを再現できるようにすることです。AI アプリにとって最も重要な記録は、モデル呼び出しとツール呼び出しの一回一回です。

何を記録するか

モジュール 03 第 4 課の call_llm はすでに記帳をしていました。呼び出しごとに 1 行の JSON を書き、時刻、モデル、所要時間、トークン、費用を含めていました。よい出発点ですが、エージェントにはまだ足りません。エージェントは一つの質問に答えるのにモデルを何回も、ツールを何回も呼ぶので、それらの記録をつなげられなければなりません。

そこで各記録にはさらに次のものが必要です。

• trace_id：同じ質問のすべての記録が一つの番号を共有します。これで一つの質問の全過程を取り出せます。
• span_id と parent_id：各記録自身の番号と、その親が誰か。たとえば「質問に答える」が親で、その下に 3 回のモデル呼び出しと 3 回のツール呼び出しがあります。
• kind と name：記録の種類（タスク、モデル呼び出し、ツール呼び出し）と具体的な名前（どのモデル、どのツール）。
• 入力と出力の要約：ツールの引数、返ってきた内容の最初の数十文字、モデルがどのツールの呼び出しを求めたか。
• 所要時間、トークン、費用、状態：成功したか、エラーになったか、それともツールがエラーメッセージを返したか。

この呼び方（trace、span）は分散システムの追跡技術から来ています。既成のオープンソースツール、たとえば Langfuse（自分でデプロイできます）や OpenTelemetry という標準も、同じ概念を使っています。まず最も簡単なものを自分で書いて、何をしているのかを理解しましょう。

数十行のトレーサー

import json
import threading
import time
import uuid
from contextlib import contextmanager
from pathlib import Path

_lock = threading.Lock()
_local = threading.local()  # 记录当前线程正在进行的 span，用来自动找到上级


class Tracer:
    def __init__(self, path):
        self.path = Path(path)

    def _write(self, record):
        with _lock, self.path.open("a") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

    @contextmanager
    def span(self, kind, name, **attrs):
        """用法：with tracer.span("llm", "chat") as s: ...; s["tokens"] = 100"""
        parent = getattr(_local, "current", None)
        record = {
            "trace_id": parent["trace_id"] if parent else uuid.uuid4().hex[:12],
            "span_id": uuid.uuid4().hex[:8],
            "parent_id": parent["span_id"] if parent else None,
            "kind": kind,
            "name": name,
            "start": time.strftime("%Y-%m-%d %H:%M:%S"),
            **attrs,
        }
        _local.current = record
        start = time.time()
        try:
            yield record
            record.setdefault("status", "ok")
        except Exception as e:
            record["status"] = "error"
            record["error"] = f"{type(e).__name__}: {e}"
            raise
        finally:
            record["ms"] = round((time.time() - start) * 1000)
            _local.current = parent
            self._write(record)

使い方は with 文です。

with tracer.span("tool", "grep_docs", args={"keyword": "timeout"}) as s:
    result = grep_docs("timeout")
    s["result_chars"] = len(result)

with ブロックが終わると、トレーサーは自動で所要時間を計算し、1 行の JSON を書きます。ブロックの中で例外が起きたらそれも記録し、状態を error にしてから、いつもどおり例外を投げます。

parent_id は自動で見つけます。トレーサーは threading.local() で「今進行中なのはどの span か」を覚えています。「質問に答える」span の中で始まったモデル呼び出しの親は、当然「質問に答える」になります。スレッドごとに自分の記録を持つので、マルチスレッドでいくつかの質問を並行して処理しても混ざりません。

エージェントにつなぐ

モジュール 05 第 2 課のエージェントのコードを変える必要はなく、外側に一枚かぶせるだけです。モデル呼び出しです。

class TracedModel(agent_loop.RealModel):
    """每次调用模型，记一个 llm 类型的 span。"""

    def __call__(self, messages, tools):
        with tracer.span("llm", self.model, input_messages=len(messages)) as s:
            message, usage = super().__call__(messages, tools)
            s.update(prompt_tokens=usage.prompt_tokens, completion_tokens=usage.completion_tokens,
                     cost=round(cost(usage), 6), tool_calls=[c.function.name for c in message.tool_calls or []])
            return message, usage

ツール呼び出しです。

def traced(name, fn):
    """包装一个工具函数，每次调用记一个 tool 类型的 span。"""
    def wrapper(**kwargs):
        with tracer.span("tool", name, args=kwargs) as s:
            result = fn(**kwargs)
            s["result_chars"] = len(result)
            s["result_preview"] = result[:80]
            if result.startswith("错误"):
                s["status"] = "tool_error"
            return result
    return wrapper


for name, t in agent_loop.TOOLS.items():
    t["fn"] = traced(name, t["fn"])

ツールが「エラー：……」を返したとき、関数そのものは例外を投げていません（モジュール 05 で扱ったとおり、エラーは観察結果にしてモデルに渡すべきだからです）。そのため別に tool_error と印を付ける必要があり、そうしないとログではすべてが正常に見えてしまいます。

一番外側では、質問ごとに task 種類の span を一つ開きます。

    for q in QUESTIONS:
        with tracer.span("task", "answer_question", question=q) as s:
            answer, stats = agent_loop.run_agent(model, q, verbose=False)
            s["answer_preview"] = (answer or "")[:60]

三つの質問を実行して、ログだけを見る

三つの質問のうち、最後のものはわざとファイル名を間違えています（正しいのは timeouts.md）。実行し終えたら、プログラムはログファイルだけを読んで分析します。後から問題を調べるときと同じです（完全なコードは code/06-production/traced_agent.py）。

日志共 25 条记录，最后一条：
{"trace_id": "173ee49bde3e", "span_id": "f9e2c3a5", "parent_id": null, "kind": "task", "name": "answer_question", "start": "2026-09-14 23:07:41", "question": "advanced/timeout.md 里写了什么？", "answer_preview": "`advanced/timeouts.md` 的内容如下（文件共 71 行；注意文件名是复数 timeouts，没有 `", "status": "ok", "ms": 5729}

「httpx 的超时分成哪几种？」 4006 毫秒，3 次模型调用，3 次工具调用，0.00104 美元
「httpx 怎么上传文件？」 5352 毫秒，3 次模型调用，6 次工具调用，0.00210 美元
「advanced/timeout.md 里写了什么？」 5729 毫秒，4 次模型调用，3 次工具调用，0.00135 美元

最慢的一步：llm deepseek-flash，3120 毫秒
工具报错次数：无

三つの質問で合計 25 件の記録です。trace_id でグループ分けすると、各質問の所要時間、呼び出し回数、費用がひと目でわかります。「ファイルをアップロードするには」はツールを 6 回呼び、他の質問の倍で、費用も倍です。その記録を開くと、files=、multipart、upload の三つのキーワードを順に検索し、さらに三つのファイルを読んだことがわかります。最も遅いステップはモデル呼び出しの一回で 3 秒あまり、ツール呼び出しはどれも数ミリ秒でした。

私はもともと、三つ目の質問では「ファイルが存在しない」というツールのエラーが一件残るだろうと思っていましたが、そうはなりませんでした。エージェントはまず list_docs を呼んでどんなファイルがあるかを見て、正しい名前が timeouts.md だとわかり、そのままそれを読んで、回答では「ファイル名は複数形なので注意」とユーザーに注意まで促していました。これもログの価値です。記録されるのは実際に起きたことで、あなたが起きるだろうと思ったことではありません。

ログで問題を調べる

このようなログがあれば、よくある問題は次のように調べられます。

JSONL ファイルは簡単で確実で、Python の数行で分析でき、個人のプロジェクトや初期の製品に向いています。量が増えたら、これらの記録を専用のツール（たとえば Langfuse）に送れば、各トレースを閲覧し、条件で検索し、統計のグラフを描くための画面が提供されます。概念は同じです。

プライバシーに注意する

ログにはユーザーの質問、ツールの引数、回答の冒頭が記録されます。ユーザーが質問に自分の携帯電話番号、パスワード、社内の情報を書いていれば、それもすべてログに入ります。

• 必要な内容だけを記録する。上のコードが記録しているのは、回答の最初の 60 文字とツールの結果の最初の 80 文字だけで、全文は記録していません。
• 機密情報を取り除く。ログに書く前に、キー、携帯電話番号、身分証番号のように見えるものを置き換えます。二つ後の課「ガードレール」で簡単な検出関数を書きます。
• 保存期間を決める。ログを永遠に残さず、たとえば 30 日だけ保存します。
• アクセス権限を管理する。ログを見られる人は、ユーザーが何を聞いたかを見られるということです。

練習問題

1. traced_agent.py を実行し、traces.jsonl を読み込んで、各質問で入力トークンが最も多かったモデル呼び出しを見つけるコードを自分で数行書いてください。
2. TracedModel にフィールドを一つ加え、呼び出しごとの入力の合計文字数を記録して、エージェントの各ステップのコンテキストがどう増えていくかを見てください。
3. わざとツールのエラーを一回起こし（たとえば一時的に read_doc を直して、あるファイルに対しては常に「エラー：……」を返すようにする）、ログの集計がそれを見つけられることを確かめてください。

確認テスト

1. trace_id、span_id、parent_id はそれぞれ何の役に立ちますか？

trace_id は同じタスク（たとえば一つの質問に答えること）のすべての記録を一つにつなぎます。span_id は各記録自身の番号です。parent_id はその親の記録を指します。この三つのフィールドがあれば、ログから一つのタスクの完全な構造を再現できます。まず何をし、その中でさらに何を呼んだのか。

2. ツールが「エラー：そのファイルはありません」を返したとき、関数は例外を投げていません。ログでどうやってそれを見つけますか？

ツールを包むコードで戻り値をチェックし、エラーメッセージだとわかったら、状態を tool_error のような値として別に印を付ける必要があります。そうしないと、ログではこの呼び出しの状態が正常になり、エラーを集計するときに見逃してしまいます。

3. ログにユーザーの質問と回答の全文を記録しないほうがよいのはなぜですか？

ユーザーの入力には個人情報、パスワード、会社の機密が含まれているかもしれず、ログが漏れたり見るべきでない人に見られたりすれば、プライバシーの問題になります。問題の調査に必要な部分だけを記録し、書き込む前に機密情報を取り除き、保存期間とアクセス権限を決めておきます。