LangSmith 追踪上线前核对：外发内容、脱敏开关与服务中断实测

"""A local LangSmith-compatible ingest endpoint that records what the SDK sends.
 
It is NOT LangSmith: it accepts the SDK's multipart run uploads, stores them, and answers with a
configurable status code and delay. It lets a lab inspect trace payloads without an account, a
network, or a real API key. Only the standard library is used; the server binds to 127.0.0.1.
"""
import email
import json
import threading
import time
from email import policy
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
# Run fields that are uploaded as separate multipart parts next to the run JSON itself.
PART_FIELDS = ("inputs", "outputs", "extra", "serialized", "events")
 
 
class IngestServer:
    def __init__(self, status: int = 200, delay: float = 0.0):
        self.status = status
        self.delay = delay
        self.posts: list[dict] = []  # one entry per upload request
        self._lock = threading.Lock()
        outer = self
 
        class Handler(BaseHTTPRequestHandler):
            def log_message(self, *args):  # keep stdout clean
                pass
 
            def _reply(self, status: int, body: bytes = b"{}", headers: dict | None = None):
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(body)))
                for key, value in (headers or {}).items():
                    self.send_header(key, value)
                self.end_headers()
                self.wfile.write(body)
 
            def do_GET(self):
                # The SDK asks for server capabilities once; an empty config selects its defaults.
                if self.path.startswith("/info"):
                    return self._reply(200, b'{"version": "local-ingest-lab"}')
                self._reply(404)
 
            def do_POST(self):
                try:
                    length = int(self.headers.get("Content-Length", 0))
                    raw = self.rfile.read(length)
                    with outer._lock:
                        outer.posts.append({"path": self.path, "ctype": self.headers.get("Content-Type", ""), "raw": raw})
                    if outer.delay:
                        time.sleep(outer.delay)
                    headers = {"Retry-After": "0"} if outer.status == 429 else None
                    self._reply(outer.status, headers=headers)
                except (BrokenPipeError, ConnectionResetError):
                    pass  # the client gave up waiting
 
        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
 
    @property
    def url(self) -> str:
        return f"http://127.0.0.1:{self.server.server_address[1]}"
 
    def runs(self) -> list[dict]:
        """Decode every upload into one record per run, merging create and update parts."""
        merged: dict[str, dict] = {}
        with self._lock:
            posts = list(self.posts)
        for post in posts:
            if not post["path"].endswith("/runs/multipart"):
                continue
            header = b"Content-Type: " + post["ctype"].encode() + b"\r\n\r\n"
            message = email.message_from_bytes(header + post["raw"], policy=policy.default)
            for part in message.iter_parts():
                name = part.get_param("name", header="content-disposition") or ""
                _, _, rest = name.partition(".")
                run_id, _, field = rest.partition(".")
                record = merged.setdefault(run_id, {})
                value = json.loads(part.get_content())
                if field:
                    record[field] = value
                else:
                    record.update(value)
        return list(merged.values())
 
    def __enter__(self):
        self.thread.start()
        return self
 
    def __exit__(self, *exc):
        self.server.shutdown()
        self.server.server_close()

"""Shared pieces: the agent under test, the sensitive markers, and helpers that read captured runs."""
import json
 
from langchain.agents import create_agent
 
from doubles import ScriptedModel, lookup_order
 
# Fake personal data. Every value is an obvious placeholder, never a real credential.
EMAIL = "[email protected]"
PHONE = "13800138000"
SECRET = "sk-test-0123456789abcdef"
USER_ID = "u-10086"
 
MARKERS = {
    "email address": EMAIL,
    "phone number": PHONE,
    "key-like string": SECRET,
    "metadata user_id": USER_ID,
}
QUESTION = f"我的手机号是{PHONE}，邮箱是{EMAIL}，密钥是{SECRET}。我的订单到哪了？"
 
 
def ask(config: dict | None = None) -> str:
    """Run the agent once and return the final answer."""
    agent = create_agent(ScriptedModel(), [lookup_order])
    merged = {"metadata": {"user_id": USER_ID}, **(config or {})}
    result = agent.invoke({"messages": [{"role": "user", "content": QUESTION}]}, config=merged)
    return result["messages"][-1].content
 
 
def fields_containing(runs: list[dict], needle: str) -> list[str]:
    """Names of the run fields in which `needle` appears, across all runs."""
    found = set()
    for run in runs:
        for field, value in run.items():
            if needle in json.dumps(value, ensure_ascii=False):
                found.add(field)
    return sorted(found)
 
 
def ordered(runs: list[dict]) -> list[dict]:
    return sorted(runs, key=lambda run: run.get("dotted_order", ""))
 
 
def depth(run: dict, by_id: dict) -> int:
    level = 0
    parent = run.get("parent_run_id")
    while parent:
        level += 1
        parent = by_id[parent].get("parent_run_id")
    return level

"""Map what one traced run sends: the run tree, how metadata spreads, and what a model run exposes."""
import json
import os
import socket
 
os.environ.setdefault("NO_PROXY", "127.0.0.1,localhost")
 
import langsmith as ls
from langchain_openai import ChatOpenAI
from langsmith import Client
 
from common import EMAIL, PHONE, USER_ID, ask, depth, ordered
from doubles import lookup_invoice
from ingest_server import IngestServer
 
PLACEHOLDER_KEY = "placeholder-api-key-0001"
 
 
def unused_url() -> str:
    """An address nobody listens on, so the model call fails fast without any provider."""
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return f"http://127.0.0.1:{sock.getsockname()[1]}/v1"
 
 
with IngestServer() as server:
    client = Client(api_url=server.url, api_key="placeholder-not-a-key", info={"version": "local-ingest-lab"})
    llm = ChatOpenAI(model="fake-model", base_url=unused_url(), api_key=PLACEHOLDER_KEY, max_retries=0, timeout=2)
    with ls.tracing_context(client=client, enabled=True):
        ask()
        try:
            lookup_invoice.invoke({"phone": PHONE})
        except ValueError:
            pass
        try:
            llm.invoke(f"hello {EMAIL}")
        except Exception:
            pass
    client.flush()
    runs = ordered(server.runs())
    uploads = sorted({post["path"] for post in server.posts})
 
by_id = {run["id"]: run for run in runs}
print(f"upload requests: {len(server.posts)} to {', '.join(uploads)}")
print(f"runs: {len(runs)} in {len({run['trace_id'] for run in runs})} traces\n")
for run in runs:
    flag = "  <- error" if run.get("error") else ""
    print(f"{'  ' * depth(run, by_id)}{run['name']} [{run['run_type']}]{flag}")
 
agent_runs = [run for run in runs if run["trace_id"] == runs[0]["trace_id"]]
carrying = sum(1 for run in agent_runs if (run["extra"].get("metadata") or {}).get("user_id") == USER_ID)
print(f"\nagent trace: runs carrying the config metadata user_id: {carrying} of {len(agent_runs)}")
print(f"extra.runtime keys: {', '.join(sorted(runs[0]['extra']['runtime']))}")
 
model = next(run for run in runs if run["name"] == "ChatOpenAI")
payload = json.dumps(model, ensure_ascii=False)
secret_ref = model["serialized"]["kwargs"]["openai_api_key"]
print("\nChatOpenAI run")
print(f"  api key value in payload: {PLACEHOLDER_KEY in payload}")
print(f"  api key field serialized as: {json.dumps(secret_ref)}")
print(f"  base url in payload: {'127.0.0.1' in json.dumps(model['serialized'])}")
print(f"  invocation_params keys: {', '.join(sorted(model['extra']['invocation_params']))}")
 
failed = next(run for run in runs if run["name"] == "lookup_invoice")
print("\nfailing tool run")
print(f"  error field holds the phone number: {PHONE in failed['error']}")
print(f"  error field holds a traceback with local file paths: {'File \"/' in failed['error']}")

upload requests: 1 to /runs/multipart
runs: 9 in 3 traces

LangGraph [chain]
  model [chain]
    ScriptedModel [llm]
  tools [chain]
    lookup_order [tool]
  model [chain]
    ScriptedModel [llm]
lookup_invoice [tool]  <- error
ChatOpenAI [llm]  <- error

agent trace: runs carrying the config metadata user_id: 7 of 7
extra.runtime keys: langchain_core_version, langchain_version, library, library_version, platform, py_implementation, runtime, runtime_version, sdk, sdk_version

ChatOpenAI run
  api key value in payload: False
  api key field serialized as: {"lc": 1, "type": "secret", "id": ["OPENAI_API_KEY"]}
  base url in payload: True
  invocation_params keys: _type, model, model_name, stop, stream

failing tool run
  error field holds the phone number: True
  error field holds a traceback with local file paths: True

"""What does each privacy switch remove from the payload that LangSmith receives?
 
Every configuration runs in its own child process because the SDK reads most environment variables
once per process. The child traces one agent run plus one failing tool call into a local ingest
server and reports which run fields still contain each fake sensitive value.
"""
import json
import os
import subprocess
import sys
 
from common import MARKERS
 
CONFIGS = [
    ("baseline", {}, {}),
    ("hide inputs (env)", {"LANGSMITH_HIDE_INPUTS": "true"}, {}),
    ("hide inputs+outputs (env)", {"LANGSMITH_HIDE_INPUTS": "true", "LANGSMITH_HIDE_OUTPUTS": "true"}, {}),
    ("hide metadata (env)", {"LANGSMITH_HIDE_METADATA": "true"}, {}),
    ("anonymizer, \\b phone regex", {}, {"phone_regex": r"\b1[3-9]\d{9}\b"}),
    ("anonymizer, lookaround regex", {}, {"phone_regex": r"(?<!\d)1[3-9]\d{9}(?!\d)"}),
    ("lookaround anonymizer + metadata", {"LANGSMITH_HIDE_METADATA": "true"}, {"phone_regex": r"(?<!\d)1[3-9]\d{9}(?!\d)"}),
    ("omit_traced_runtime_info", {}, {"omit_runtime": True}),
    ("tracing_context(enabled=False)", {}, {"enabled": False}),
]
SHORT = {"inputs": "in", "outputs": "out", "extra": "meta", "error": "err", "serialized": "ser"}
ORDER = ["inputs", "outputs", "extra", "error", "serialized"]
 
 
def child(index: int) -> None:
    import langsmith as ls
    from langsmith import Client
    from langsmith.anonymizer import create_anonymizer
 
    from common import PHONE, ask, fields_containing
    from doubles import lookup_invoice
    from ingest_server import IngestServer
 
    _, _, options = CONFIGS[index]
    patterns = [
        {"pattern": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", "replace": "<email>"},
        {"pattern": options.get("phone_regex", ""), "replace": "<phone>"},
        {"pattern": r"sk-[A-Za-z0-9-]{10,}", "replace": "<key>"},
    ]
    with IngestServer() as server:
        client = Client(
            api_url=server.url,
            api_key="placeholder-not-a-key",
            info={"version": "local-ingest-lab"},
            anonymizer=create_anonymizer(patterns) if "phone_regex" in options else None,
            omit_traced_runtime_info=bool(options.get("omit_runtime")),
        )
        with ls.tracing_context(client=client, enabled=options.get("enabled", True)):
            ask()
            try:
                lookup_invoice.invoke({"phone": PHONE})
            except ValueError:
                pass
        client.flush()
        runs = server.runs()
    cells = {name: fields_containing(runs, value) for name, value in MARKERS.items()}
    runtime = any("runtime" in (run.get("extra") or {}) for run in runs)
    print(json.dumps({"runs": len(runs), "cells": cells, "runtime": runtime}))
 
 
def cell(fields: list[str]) -> str:
    return "+".join(SHORT[f] for f in ORDER if f in fields) or "-"
 
 
def main() -> None:
    names = list(MARKERS)
    print(f"{'configuration':<34}{'runs':>5}  " + "  ".join(f"{n:<17}" for n in names) + "  runtime")
    for index, (label, env, _) in enumerate(CONFIGS):
        child_env = {k: v for k, v in os.environ.items() if not k.startswith("LANGSMITH_")}
        child_env.update(env)
        done = subprocess.run(
            [sys.executable, __file__, "--child", str(index)], env=child_env, capture_output=True, text=True, timeout=60
        )
        if done.returncode != 0:
            raise SystemExit(f"child failed for {label!r}:\n{done.stderr[-600:]}")
        result = json.loads(done.stdout.strip().splitlines()[-1])
        row = "  ".join(f"{cell(result['cells'][n]):<17}" for n in names)
        print(f"{label:<34}{result['runs']:>5}  {row}  {'yes' if result['runtime'] else 'no'}")
 
 
if __name__ == "__main__":
    if len(sys.argv) == 3 and sys.argv[1] == "--child":
        child(int(sys.argv[2]))
    else:
        main()

configuration                      runs  email address      phone number       key-like string    metadata user_id   runtime
baseline                              8  in+out             in+out+err         in+out             meta               yes
hide inputs (env)                     8  out                out+err            out                meta               yes
hide inputs+outputs (env)             8  -                  err                -                  meta               yes
hide metadata (env)                   8  in+out             in+out+err         in+out             -                  yes
anonymizer, \b phone regex            8  -                  in+out+err         -                  meta               yes
anonymizer, lookaround regex          8  -                  -                  -                  meta               yes
lookaround anonymizer + metadata      8  -                  -                  -                  -                  yes
omit_traced_runtime_info              8  in+out             in+out+err         in+out             meta               yes
tracing_context(enabled=False)        0  -                  -                  -                  -                  no

"""What happens to the application when the trace endpoint is unavailable or slow?"""
import socket
import time
 
import langsmith as ls
from langsmith import Client
 
from common import ask
from ingest_server import IngestServer
 
 
def unused_url() -> str:
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return f"http://127.0.0.1:{sock.getsockname()[1]}"
 
 
def bucket(seconds: float) -> str:
    return "<1s" if seconds < 1 else "1-5s" if seconds <= 5 else ">5s"
 
 
SCENARIOS = [
    ("endpoint healthy", {}),
    ("connection refused", {"refused": True}),
    ("HTTP 500 on every upload", {"status": 500}),
    ("HTTP 429 on every upload", {"status": 429}),
    ("HTTP 401 (bad key)", {"status": 401}),
    ("endpoint answers after 3 s", {"delay": 3.0}),
]
 
print(f"{'scenario':<28}{'answer ok':<11}{'app call':<10}{'flush':<8}{'uploads':<9}error callback")
for label, options in SCENARIOS:
    errors: list[str] = []
    with IngestServer(status=options.get("status", 200), delay=options.get("delay", 0.0)) as server:
        client = Client(
            api_url=unused_url() if options.get("refused") else server.url,
            api_key="placeholder-not-a-key",
            info={"version": "local-ingest-lab"},
            tracing_error_callback=lambda exc: errors.append(type(exc).__name__),
        )
        started = time.perf_counter()
        with ls.tracing_context(client=client, enabled=True):
            answer = ask()
        called = time.perf_counter()
        client.flush()
        flushed = time.perf_counter()
    callback = ", ".join(sorted(set(errors))) or "-"
    ok = "yes" if answer.startswith("订单 A-1001") else "NO"
    print(f"{label:<28}{ok:<11}{bucket(called - started):<10}{bucket(flushed - called):<8}{len(server.posts):<9}{callback}")

scenario                    answer ok  app call  flush   uploads  error callback
endpoint healthy            yes        <1s       <1s     1        -
connection refused          yes        <1s       >5s     0        LangSmithConnectionError
HTTP 500 on every upload    yes        <1s       >5s     12       LangSmithAPIError
HTTP 429 on every upload    yes        <1s       1-5s    4        LangSmithRateLimitError
HTTP 401 (bad key)          yes        <1s       <1s     1        LangSmithAuthError
endpoint answers after 3 s  yes        <1s       1-5s    1        -

"""Does trace sampling keep or drop whole traces, and does the rate behave as configured?"""
import langsmith as ls
from langsmith import Client
 
from common import ask
from ingest_server import IngestServer
 
ASKS = 100
RUNS_PER_TRACE = 7  # one agent run produces seven runs (see egress_map.py)
 
 
def trial(rate: float) -> dict:
    with IngestServer() as server:
        client = Client(
            api_url=server.url,
            api_key="placeholder-not-a-key",
            info={"version": "local-ingest-lab"},
            tracing_sampling_rate=rate,
        )
        with ls.tracing_context(client=client, enabled=True):
            for _ in range(ASKS):
                ask()
        client.flush()
        runs = server.runs()
    traces: dict[str, int] = {}
    for run in runs:
        traces[run["trace_id"]] = traces.get(run["trace_id"], 0) + 1
    return {"traces": len(traces), "whole": all(count == RUNS_PER_TRACE for count in traces.values())}
 
 
print(f"{ASKS} agent calls per setting")
zero = trial(0.0)
print(f"rate 0.0: traces kept == 0: {zero['traces'] == 0}")
full = trial(1.0)
print(f"rate 1.0: traces kept == {ASKS}: {full['traces'] == ASKS}")
half = trial(0.5)
print(f"rate 0.5: traces kept within 25..75: {25 <= half['traces'] <= 75}")
print(f"rate 0.5: every kept trace has all {RUNS_PER_TRACE} runs: {half['whole']}")

100 agent calls per setting
rate 0.0: traces kept == 0: True
rate 1.0: traces kept == 100: True
rate 0.5: traces kept within 25..75: True
rate 0.5: every kept trace has all 7 runs: True