Multi-Agent Workflow with Observability

This is the shape of production tracing with LangSmith — a preview of what you'll build by hand in the labs below. The big idea: you rarely write logging code by hand; you add a decorator and the library records each call for you.

1. from langsmith import traceable pulls in the decorator. A decorator is a line starting with @ placed just above a function; it wraps that function to add behaviour — here, recording a trace — without changing the function's own code.
2. @traceable(run_type="chain") on research_agent means: every time this agent runs, capture a span (one entry in the trace). Because retrieve and llm_answer are called inside it, their spans nest underneath automatically — that's the tree you inspect later.
3. The second function, score_run, is an online eval: it takes a run's inputs and outputs and returns quality numbers — here a "faithfulness" judge score and the run's "cost_usd". Running this on live traffic is how you catch quality drift.

Try this: Notice you didn't write any 'save to database' code — the @traceable decorator does the capturing. The rest of this page rebuilds this exact idea in ~30 lines of plain Python so you can see what the decorator is really doing.

This file is the heart of the project: a tiny tracer. It defines a @traced decorator that records one span (a little dictionary describing a single step) every time a wrapped function runs — and, crucially, even when that function crashes.

1. RUN = {"spans": []} is a shared list where every span gets appended. PII = re.compile(...) is a pattern that matches things that look like a Social Security number (123-45-6789) or a 16-digit card number.
2. redact(value) turns any value into text and replaces anything matching that PII pattern with "[REDACTED]" — so customer secrets never get stored in a trace.
3. traced(name, tokens=0) is the decorator. When you write @traced("planner", tokens=300) above a function, deco then wrap replace it with a version that does bookkeeping around the real call. @functools.wraps(fn) just keeps the original function's name/docs.
4. Inside wrap: it builds the span (recording the redacted input and token count), starts a timer with time.perf_counter(), then runs the real function in a try. On success it saves the redacted output; on failure it saves the error message and re-raises.
5. The finally block runs no matter what — success or crash — recording the elapsed ms and appending the span to RUN. That guarantee is the whole trick (next box).

What the output means: Nothing prints yet — this file only defines tools. After a workflow runs, RUN["spans"] holds one dictionary per step, each with name, input, output, error, tokens, and ms.

Try this: Read the try / except / finally shape slowly: try attempts the work, except catches a crash, and finally always runs the cleanup. Here 'cleanup' means 'record the span' — which is why even failures show up in the trace.

Raw spans are detailed but hard to eyeball. This file rolls them up into the four numbers an operator actually watches — step count, total time, error count, and estimated cost — plus a helper to name the step that failed.

1. PRICE_PER_TOKEN = 0.00001 is a made-up price so we can turn token counts into dollars. Real prices differ per model, but the math is identical.
2. rollup(run) reads run["spans"] and returns a summary dictionary. len(spans) is the number of steps; sum(s["ms"] for s in spans) adds every step's time; sum(1 for s in spans if s["error"]) counts how many steps recorded an error.
3. cost_usd multiplies the total tokens across all spans by the price per token. round(..., 2) and round(..., 4) just trim the numbers to a readable length.
4. failing_step(run) walks the spans in order and returns the name of the first one whose error isn't empty, or None if nothing failed. This is the one-line answer to 'which step broke?'.

What the output means: rollup gives you something like {'steps': 4, 'total_ms': 0.02, 'errors': 0, 'cost_usd': 0.0322}; failing_step gives a step name (e.g. "researcher") or None.

Try this: These are ordinary Python comprehensions — sum(... for s in spans if ...) reads as 'for each span, maybe add this'. Change the if s["error"] to count something else, like steps over 100ms, and you've invented a new metric.

Now we put the tracer to work. This is a pretend multi-agent 'crew' — a planner, a researcher, and a writer — with each step wrapped in @traced. No language model is needed; the functions return canned strings so you can focus purely on what observability shows you.

1. from trace import traced, RUN, reset and from metrics import rollup, failing_step pull in the tools you just built. This is why the files must live in the same folder.
2. Each agent is a normal function with a @traced(...) decorator giving it a name and a token count: planner (300 tokens) returns two sub-questions, researcher (1200) answers one, writer (500) joins the findings into a report. The big token number on researcher is a hint that research will dominate cost.
3. run(question) is the workflow: it calls reset() to clear old spans, gets sub-questions from the planner, runs the researcher on each one (a list comprehension), then writes the report. Because every step is traced, RUN["spans"] fills up automatically as this runs.
4. The if __name__ == "__main__": block runs only when you execute the file directly. It prints the report, then loops over RUN["spans"] to print a readable run tree (name, ms, tokens, error per step), then prints the rollup metrics.

What the output means: You get the final report string, a line-per-step run tree, and the metrics summary — see the terminal box just below.

Try this: The f"{s['name']:12}" bits are f-string alignment: :12 pads the name to 12 characters so the columns line up. Change researcher's tokens and watch cost_usd move in the metrics.

This is what python workflow.py actually prints. It's the first time you see observability paying off: the run tree and metrics were produced automatically, just because each step was @traced.

1. The first line is the workflow's real output: the joined report.
2. Under --- run tree ---, each line is one span: the step name, how long it took in ms, its token count, and its error (None = no error). Notice researcher appears twice — once per sub-question — which matches the loop in run().
3. Under --- metrics ---, the rollup dictionary summarises the whole run: 4 steps, near-zero total time (the functions do no real work), 0 errors, and $0.0322 estimated cost.

What the output means: Times are tiny here because the fake agents return instantly; in a real system those numbers would be seconds and dollars, and this same tree would show you exactly where they went.

Try this: Add a fourth agent (or a second writer call) and re-run: the run tree grows a line and the metrics update on their own — you never touched the printing or metrics code.

These tests prove the tracer behaves — and they run with no API key and no network, because the whole tracer is plain Python. Each test is a small, focused claim about one behaviour.

1. setup_function() runs before every test and calls reset(), so spans from one test can't leak into the next — a common source of confusing failures.
2. test_span_records_io_and_ms wraps a trivial function, calls it, and checks the recorded span has the right name and a timing field. test_error_span_is_captured makes a function that raises, wraps the call in pytest.raises(ValueError) (which says 'we expect this to throw'), then asserts the error was still recorded — proving the finally works.
3. test_rollup_counts_and_costs calls a traced function twice and checks the metrics say 2 steps, 0 errors, and a positive cost. test_pii_redacted_before_store passes in a fake SSN and asserts "[REDACTED]" ended up in the stored input.
4. test_failing_step_is_findable runs one good step and one bad step and asserts failing_step(RUN) returns "bad" — the payoff of the whole project. test_redact_leaves_clean_text confirms redaction doesn't mangle ordinary text.

What the output means: Run with python -m pytest tests/ -v you should see all 6 passed — each line is one of the behaviours above proven true.

Try this: Comment out the finally block in trace.py and re-run: test_error_span_is_captured should fail, because the crashing step's span never gets recorded. That failing test is the safety net working.

The finale: because your @traced decorator has the same shape as LangSmith's real @traceable, going from your homemade tracer to a hosted, production dashboard is essentially a one-line change.

1. In the terminal, pip install langsmith adds the real library and export LANGSMITH_API_KEY=... gives it credentials (like the API keys used elsewhere in the course).
2. In workflow.py, you swap from trace import traced for from langsmith import traceable, and change the decorator to @traceable(run_type="chain"). The function body doesn't change at all.
3. Now the same run you saw in the terminal is captured to LangSmith's hosted UI, where the trace tree, metrics, and evals are shown as interactive dashboards — no dashboard code to write.

Try this: Keep your local @traced for the tests (fast and free) and use @traceable only in the running app. That split — cheap local tracing for CI, hosted tracing for production — is exactly how real teams operate.