Registrar llamadas a herramientas y entradas/salidas

Qué registrar en cada paso

Auxiliar de spans personalizado

from contextlib import contextmanager
import time, uuid

@contextmanager
def span(name, trace_id, parent_id=None):
    span_id = str(uuid.uuid4())
    start = time.time()
    record = {'trace_id': trace_id, 'span_id': span_id, 'parent_id': parent_id, 'name': name, 'start': start}
    try:
        yield record
    except Exception as e:
        record['error'] = str(e)
        raise
    finally:
        record['duration_ms'] = (time.time() - start) * 1000
        save_span(record)

Using It

trace_id = str(uuid.uuid4())
with span('llm_call_planner', trace_id) as s:
    s['model'] = 'gpt-4o-mini'
    s['messages_in'] = messages
    response = client.chat.completions.create(model='gpt-4o-mini', messages=messages)
    s['response'] = response.choices[0].message.content
    s['tokens_in'] = response.usage.prompt_tokens
    s['tokens_out'] = response.usage.completion_tokens

Anidamiento de spans

with span('agent_step', trace_id) as parent:
    with span('llm_call', trace_id, parent_id=parent['span_id']):
        ...
    with span('tool_call:search', trace_id, parent_id=parent['span_id']):
        ...

Registro de llamadas a herramientas

Truncar cargas grandes

def safe_payload(obj, max_chars=8000):
    s = json.dumps(obj)
    return s if len(s) <= max_chars else s[:max_chars] + '...[truncated]'

Ocultación de PII

import re
EMAIL_RE = re.compile(r'[\w\.-]+@[\w\.-]+')

def redact(text):
    return EMAIL_RE.sub('[email]', text)

print(redact("Contact me at alice@example.com for details."))

Salida de registros estructurados

import json
import sys

def save_span(record):
    print(json.dumps(record), file=sys.stderr, flush=True)

demo_record = {'span': 'call_llm', 'duration_ms': 120, 'ok': True}
save_span(demo_record)
print("Logged span:", json.dumps(demo_record))

Correlacione con las sesiones de usuario

Muestreo

import random

def should_trace(user_id):
    return random.random() < 0.10  # 10% of traces
# Always trace errors and slow runs.

random.seed(42)
sampled = sum(should_trace(i) for i in range(1000))
print(f"Traced {sampled} out of 1000 calls (~10% target)")

Reproduzca a partir de las trazas

¿Debe truncar las cargas útiles grandes?

Resumen