L2 · 加 4 件工程化 · trace / checkpoint / middleware / subagent

① Tracing · 看见 agent 在干啥

最少代码 · ~30 行的 minimal tracer

"""minimal tracer · stdout JSONL · OTel-compatible field names."""
import json, os, time, uuid
from contextvars import ContextVar
from contextlib import contextmanager

_current_trace: ContextVar[str | None] = ContextVar("trace", default=None)
_current_span:  ContextVar[str | None] = ContextVar("span", default=None)

LOG_PATH = os.environ.get("AGENT_TRACE_LOG", "trace.jsonl")

def _emit(event: dict): open(LOG_PATH, "a").write(json.dumps(event) + "\n")

@contextmanager
def trace_run(name: str, **attrs):
    tid = uuid.uuid4().hex[:12]
    tok = _current_trace.set(tid)
    t0 = time.time()
    _emit({"kind": "trace_start", "trace_id": tid, "name": name, "ts": t0, **attrs})
    try: yield tid
    finally:
        _emit({"kind": "trace_end", "trace_id": tid, "duration_ms": int((time.time() - t0) * 1000)})
        _current_trace.reset(tok)

@contextmanager
def span(name: str, **attrs):
    sid = uuid.uuid4().hex[:8]
    parent = _current_span.get()
    tok = _current_span.set(sid)
    t0 = time.time()
    _emit({"kind": "span_start", "trace_id": _current_trace.get(), "span_id": sid,
           "parent_span_id": parent, "name": name, "ts": t0, **attrs})
    try:
        result = (yield)
        _emit({"kind": "span_end", "span_id": sid,
               "duration_ms": int((time.time() - t0) * 1000), "status": "ok"})
    except Exception as e:
        _emit({"kind": "span_end", "span_id": sid,
               "status": "error", "error": repr(e)})
        raise
    finally: _current_span.reset(tok)

diff 到 minimal_agent.py

from tracer import trace_run, span     # +import

def run_agent(user_input, *, max_steps=8, model="openai/gpt-4o-mini"):
    with trace_run("agent_run", model=model, query=user_input):       # +wrap
        client = OpenAI(...)
        history = [Step("user", user_input)]

        for step_no in range(max_steps):
            with span("step", step_no=step_no):                          # +wrap
                with span("llm_call", model=model):                       # +wrap
                    rsp = client.chat.completions.create(...)
                msg = rsp.choices[0].message

                if not msg.tool_calls:
                    history.append(Step("assistant", msg.content or ""))
                    return msg.content or ""

                for tc in msg.tool_calls:
                    with span("tool_call", name=tc.function.name):           # +wrap
                        result = TOOLS[tc.function.name](**json.loads(tc.function.arguments))
                        history.append(Step("tool", result, tool_call_id=tc.id))

$ python minimal_agent.py
$ jq -c '.' trace.jsonl | head -8

{"kind":"trace_start","trace_id":"a1b2c3d4e5f6","name":"agent_run","model":"openai/gpt-4o-mini","query":"sf weather?",...}
{"kind":"span_start","trace_id":"a1b2...","span_id":"01a2","parent_span_id":null,"name":"step","step_no":0,...}
{"kind":"span_start","trace_id":"a1b2...","span_id":"01a3","parent_span_id":"01a2","name":"llm_call",...}
{"kind":"span_end","span_id":"01a3","duration_ms":842,"status":"ok"}
{"kind":"span_start","trace_id":"a1b2...","span_id":"01a4","parent_span_id":"01a2","name":"tool_call","name":"get_weather",...}
{"kind":"span_end","span_id":"01a4","duration_ms":2,"status":"ok"}
{"kind":"span_end","span_id":"01a2","duration_ms":848,"status":"ok"}
{"kind":"trace_end","trace_id":"a1b2c3d4e5f6","duration_ms":1230}