Logging and observability

Once the application is live, users use it where you can't see. One day someone reports "your assistant gave a completely wrong answer", or the monthly bill is three times what you expected. How do you investigate?

If all you have is the user's question and the final answer, you have almost nothing to go on. Did retrieval miss the right document? Did the model misread it? Did a tool fail? Or did it go astray at some step and get further and further off from there?

Observability means the system leaves enough records while it runs that you can reconstruct afterwards what it actually did. For AI applications, the most important records are every model call and every tool call.

What to record

Module 03, Lesson 4's call_llm already keeps accounts: one line of JSON per call, with the time, model, duration, tokens and cost. That's a good start, but not enough for an agent. An agent calls the model several times and tools several times to answer one question, and those records need to be linked together.

So each record also needs:

• trace_id: all records for the same question share one id. With it you can pull out a question's entire process.
• span_id and parent_id: each record's own id, and the id of its parent. For example, "answer the question" is the parent, with three model calls and three tool calls under it.
• kind and name: the type of record (task, model call, tool call) and its specific name (which model, which tool).
• Summaries of input and output: tool arguments, the first few dozen characters of what came back, which tools the model asked to call.
• Duration, tokens, cost, status: success, error, or a tool that returned an error message.

These terms (trace, span) come from tracing in distributed systems. Ready-made open-source tools such as Langfuse (which you can self-host), and the OpenTelemetry standard, all use the same concepts. We first write the simplest possible one ourselves to understand what it does.

A tracer in a few dozen lines

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)

You use it with a with statement:

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

When the with block ends, the tracer computes the duration and writes one line of JSON. If an exception is raised inside the block, it's recorded too, with status error, and the exception is then raised as usual.

parent_id is found automatically: the tracer uses threading.local() to remember "which span is currently in progress". A model call started inside the "answer the question" span naturally has "answer the question" as its parent. Each thread has its own record, so handling several questions concurrently on multiple threads doesn't mix them up.

Hooking it up to the agent

There's no need to change the agent code from Module 05, Lesson 2; just wrap it. Model calls:

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

Tool calls:

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"])

When a tool returns "Error: …", the function itself hasn't raised an exception (as Module 05 explained, errors become observations handed to the model), so it has to be marked tool_error separately; otherwise the log makes everything look normal.

At the outermost level, each question opens a span of kind task:

    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]

Run three questions, then look only at the logs

Three questions; the last one has a deliberately misspelled file name (the right one is timeouts.md). After the run, the program reads only the log file to analyse it, just as you would when investigating after the fact (full code in 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 毫秒
工具报错次数：无

Three questions produced 25 records in total. Grouped by trace_id, each question's duration, number of calls and cost are clear at a glance. "How do I upload a file" made 6 tool calls, twice as many as the others, and cost twice as much; open its records and you can see it searched for files=, multipart and upload in turn, and read three files. The slowest step was a model call of just over 3 seconds; tool calls took only milliseconds.

I had expected the third question to leave a "file not found" tool error, but it didn't: the agent first called list_docs to see what files exist, found the right name was timeouts.md, read it directly, and even reminded the user in its answer that "the file name is plural". That's another value of logs: they record what actually happened, not what you assumed would happen.

Using logs to investigate

With logs like these, common problems are investigated like this:

A JSONL file is simple and reliable, and a few lines of Python can analyse it, which suits personal projects and early products. At larger volumes, you can send these records to a dedicated tool (such as Langfuse), which provides an interface for browsing each trace, searching by conditions and drawing charts. The concepts are the same.

Mind privacy

The logs record users' questions, tool arguments and the start of answers. If a user writes their phone number, a password or internal company information in a question, all of it goes into the logs.

• Record only what's necessary. The code above records only the first 60 characters of the answer and the first 80 of tool results, not the full content.
• Filter sensitive information. Before writing to the log, replace anything that looks like a key, a phone number or an ID number. The "guardrails" lesson after next writes a simple detection function.
• Set a retention period. Don't keep logs forever; keep them for 30 days, say.
• Control access. Anyone who can read the logs can see what users asked.

Exercises

1. Run traced_agent.py, then write a few lines of your own to read traces.jsonl and find, for each question, the model call with the most input tokens.
2. Add a field to TracedModel that records the total number of input characters on each call, and see how the agent's context grows from step to step.
3. Deliberately cause a tool error (for example, temporarily change read_doc to always return "Error: …" for one file) and confirm the log statistics catch it.

Self-check

1. What are trace_id, span_id and parent_id each for?

trace_id links together all records of the same task (such as answering one question); span_id is each record's own id; parent_id points to its parent record. With these three fields you can reconstruct a task's full structure from the logs: what it did first, and what that in turn called.

2. When a tool returns "Error: no such file", the function didn't raise an exception. How can the logs catch it?

The code wrapping the tool has to check the return value, and when it's an error message, mark the status separately as something like tool_error. Otherwise the call's status in the log is normal, and counts of errors will miss it.

3. Why shouldn't the logs record users' full questions and answers?

Users' input may contain personal information, passwords or company secrets, and if the logs leak or are seen by the wrong people, that's a privacy problem. Record only what's needed for investigating, filter sensitive information before writing, and set a retention period and access controls.