# Multi-Agent PR Review Pipeline with Persistent State and OTel Tracing > Build a four-stage code-review pipeline in LangGraph that stores findings as JSON artifacts between stages, instruments each sub-agent call with OpenTelemetry spans, and surfaces per-stage latency and token counts in a local console trace backend, no external services required. - Canonical URL: https://agentry.press/tutorial/multi-agent-pr-review-pipeline-with-persistent-state-and-otel-tracing/ - Type: Tutorial - Published: 2026-05-25 - By: agentry - Tags: langgraph, multi-agent, opentelemetry, code-review, tracing, python --- ## Why this matters The adamsreview project [1] demonstrated a concrete engineering pattern: break a monolithic AI code review into discrete stages, persist findings as JSON artifacts between them so context windows stay clean, and run validation passes that can revert regressions. The key insight is that clearing context between stages is only safe when state is durable. Without that durability, a multi-stage agent either stuffs every prior finding into every subsequent prompt (expensive, noisy) or loses findings entirely. For operators running these pipelines in CI, a second problem emerges: when a review takes 45 seconds instead of 12, you cannot tell whether the regression is in the diff-parsing stage, the security-analysis stage, or the synthesis stage. Without per-stage spans, you're debugging a black box. This tutorial wires OpenTelemetry spans onto each LangGraph node so latency and token usage are visible per stage, using only a local console exporter that requires no credentials. ## Prerequisites - Python 3.11 or 3.12 - An Anthropic API key (`ANTHROPIC_API_KEY`) or OpenAI API key (`OPENAI_API_KEY`) - Basic familiarity with LangGraph nodes and state graphs - A small Git diff or code snippet to review (the tutorial ships a built-in sample) ## Setup Install the required packages. The pipeline uses LangGraph for orchestration, `langchain-anthropic` for the LLM calls, and the OpenTelemetry SDK with its console exporter for local tracing. ```bash uv pip install langgraph langchain-anthropic opentelemetry-sdk opentelemetry-api ``` Export your API key. The blocks that call the Anthropic API are marked `skip_execution_reason` because the sandbox has no key forwarded. Every other block runs. ```bash export ANTHROPIC_API_KEY="sk-ant-your-key-here" ``` ## Step 1: Define the shared review state The pipeline passes a single `ReviewState` TypedDict through every node. Findings from each stage accumulate in a list; the JSON artifact path is written once and read by later stages. This mirrors the adamsreview pattern [1] of storing state in JSON artifacts on disk so context can be cleared between stages. ```python # filename: review_state.py from __future__ import annotations from typing import TypedDict, Annotated import operator class StageResult(TypedDict): stage: str findings: list[str] token_usage: dict latency_ms: float class ReviewState(TypedDict): diff: str # raw diff text artifact_path: str # path to on-disk JSON stage_results: Annotated[list[StageResult], operator.add] final_report: str ``` ## Step 2: Set up the OTel tracer Configure a `TracerProvider` with a `SimpleSpanProcessor` backed by a `ConsoleSpanExporter`. The `SimpleSpanProcessor` flushes each span synchronously the moment it ends, which means spans are visible immediately in the same process rather than buffered until exit. ```python # filename: tracing.py from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter from opentelemetry.sdk.resources import Resource def configure_tracer(service_name: str = "pr-review-pipeline") -> trace.Tracer: resource = Resource.create({"service.name": service_name}) provider = TracerProvider(resource=resource) provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter())) trace.set_tracer_provider(provider) return trace.get_tracer(service_name) tracer = configure_tracer() ``` ## Step 3: Build the JSON artifact helpers Each stage reads the current artifact, appends its findings, and writes the updated artifact back. This is the persistence layer: if the pipeline crashes mid-run, completed stages are not lost. ```python # filename: artifact_store.py import json import os import time from pathlib import Path ARTIFACT_DIR = Path("/tmp/review_artifacts") ARTIFACT_DIR.mkdir(exist_ok=True) def init_artifact(run_id: str) -> str: path = str(ARTIFACT_DIR / f"{run_id}.json") payload = {"run_id": run_id, "created_at": time.time(), "stages": {}} Path(path).write_text(json.dumps(payload, indent=2)) return path def append_stage(artifact_path: str, stage: str, findings: list[str]) -> None: data = json.loads(Path(artifact_path).read_text()) data["stages"][stage] = {"findings": findings, "written_at": time.time()} Path(artifact_path).write_text(json.dumps(data, indent=2)) def read_artifact(artifact_path: str) -> dict: return json.loads(Path(artifact_path).read_text()) ``` ## Step 4: Implement the four review stages Each stage is a plain Python function that becomes a LangGraph node. The pattern is identical across stages: open a span, call the LLM, record token usage as span attributes, write findings to the artifact, and return the state delta. The `_call_llm` helper is a thin wrapper that works with either `langchain-anthropic` or a mock. In the runnable sandbox blocks, a mock replaces the real LLM call so the pipeline executes without an API key. ```python # filename: stages.py from __future__ import annotations import time from typing import Any from artifact_store import append_stage from review_state import ReviewState, StageResult from tracing import tracer # --------------------------------------------------------------------------- # Thin LLM abstraction — swap in real langchain-anthropic at runtime # --------------------------------------------------------------------------- def _call_llm(system: str, user: str, llm: Any | None = None) -> tuple[str, dict]: """Returns (response_text, token_usage_dict).""" if llm is None: # Mock path used in tests / sandbox mock_text = f"[MOCK] Reviewed with system='{system[:40]}'" return mock_text, {"input_tokens": 0, "output_tokens": 0} # Real path from langchain_core.messages import HumanMessage, SystemMessage response = llm.invoke([SystemMessage(content=system), HumanMessage(content=user)]) usage = getattr(response, "usage_metadata", {}) or {} return response.content, { "input_tokens": usage.get("input_tokens", 0), "output_tokens": usage.get("output_tokens", 0), } # --------------------------------------------------------------------------- # Stage 1: Parse and summarise the diff # --------------------------------------------------------------------------- def stage_parse_diff(state: ReviewState, llm: Any = None) -> dict: with tracer.start_as_current_span("stage.parse_diff") as span: t0 = time.perf_counter() span.set_attribute("review.stage", "parse_diff") span.set_attribute("diff.length_chars", len(state["diff"])) system = "You are a diff parser. Summarise what changed: files, functions, and intent." text, usage = _call_llm(system, state["diff"], llm) latency = (time.perf_counter() - t0) * 1000 span.set_attribute("llm.input_tokens", usage["input_tokens"]) span.set_attribute("llm.output_tokens", usage["output_tokens"]) span.set_attribute("stage.latency_ms", round(latency, 2)) findings = [text] append_stage(state["artifact_path"], "parse_diff", findings) result: StageResult = { "stage": "parse_diff", "findings": findings, "token_usage": usage, "latency_ms": round(latency, 2), } return {"stage_results": [result]} # --------------------------------------------------------------------------- # Stage 2: Security analysis # --------------------------------------------------------------------------- def stage_security(state: ReviewState, llm: Any = None) -> dict: with tracer.start_as_current_span("stage.security") as span: t0 = time.perf_counter() span.set_attribute("review.stage", "security") prior = state["stage_results"][-1]["findings"][0] if state["stage_results"] else "" system = ( "You are a security reviewer. Identify injection risks, secret leakage, " "insecure defaults, and authentication bypasses in the diff." ) user = f"Diff summary:\n{prior}\n\nFull diff:\n{state['diff']}" text, usage = _call_llm(system, user, llm) latency = (time.perf_counter() - t0) * 1000 span.set_attribute("llm.input_tokens", usage["input_tokens"]) span.set_attribute("llm.output_tokens", usage["output_tokens"]) span.set_attribute("stage.latency_ms", round(latency, 2)) findings = [text] append_stage(state["artifact_path"], "security", findings) result: StageResult = { "stage": "security", "findings": findings, "token_usage": usage, "latency_ms": round(latency, 2), } return {"stage_results": [result]} # --------------------------------------------------------------------------- # Stage 3: Logic and correctness review # --------------------------------------------------------------------------- def stage_logic(state: ReviewState, llm: Any = None) -> dict: with tracer.start_as_current_span("stage.logic") as span: t0 = time.perf_counter() span.set_attribute("review.stage", "logic") system = ( "You are a logic reviewer. Find off-by-one errors, incorrect conditionals, " "missing error handling, and algorithmic bugs." ) user = f"Diff:\n{state['diff']}" text, usage = _call_llm(system, user, llm) latency = (time.perf_counter() - t0) * 1000 span.set_attribute("llm.input_tokens", usage["input_tokens"]) span.set_attribute("llm.output_tokens", usage["output_tokens"]) span.set_attribute("stage.latency_ms", round(latency, 2)) findings = [text] append_stage(state["artifact_path"], "logic", findings) result: StageResult = { "stage": "logic", "findings": findings, "token_usage": usage, "latency_ms": round(latency, 2), } return {"stage_results": [result]} # --------------------------------------------------------------------------- # Stage 4: Synthesis — produce the final report # --------------------------------------------------------------------------- def stage_synthesise(state: ReviewState, llm: Any = None) -> dict: with tracer.start_as_current_span("stage.synthesise") as span: t0 = time.perf_counter() span.set_attribute("review.stage", "synthesise") all_findings = "\n\n".join( f"[{r['stage']}]\n" + "\n".join(r["findings"]) for r in state["stage_results"] ) system = ( "You are a senior engineer writing a final PR review. " "Consolidate the findings below into a concise, prioritised report. " "Group by severity: CRITICAL, HIGH, MEDIUM, LOW." ) text, usage = _call_llm(system, all_findings, llm) latency = (time.perf_counter() - t0) * 1000 span.set_attribute("llm.input_tokens", usage["input_tokens"]) span.set_attribute("llm.output_tokens", usage["output_tokens"]) span.set_attribute("stage.latency_ms", round(latency, 2)) append_stage(state["artifact_path"], "synthesise", [text]) return {"final_report": text} ``` ## Step 5: Wire the LangGraph pipeline Connect the four stages into a `StateGraph`. Each node is a lambda that closes over the `llm` argument so the graph itself stays stateless and testable. ```python # filename: pipeline.py from __future__ import annotations from typing import Any import uuid from langgraph.graph import StateGraph, END from artifact_store import init_artifact from review_state import ReviewState from stages import stage_parse_diff, stage_security, stage_logic, stage_synthesise from tracing import tracer def build_pipeline(llm: Any = None): """Return a compiled LangGraph review pipeline.""" def _parse(state: ReviewState) -> dict: return stage_parse_diff(state, llm) def _security(state: ReviewState) -> dict: return stage_security(state, llm) def _logic(state: ReviewState) -> dict: return stage_logic(state, llm) def _synthesise(state: ReviewState) -> dict: return stage_synthesise(state, llm) graph = StateGraph(ReviewState) graph.add_node("parse_diff", _parse) graph.add_node("security", _security) graph.add_node("logic", _logic) graph.add_node("synthesise", _synthesise) graph.set_entry_point("parse_diff") graph.add_edge("parse_diff", "security") graph.add_edge("security", "logic") graph.add_edge("logic", "synthesise") graph.add_edge("synthesise", END) return graph.compile() def run_review(diff: str, llm: Any = None) -> ReviewState: """Run a full review and return the final state.""" run_id = uuid.uuid4().hex[:8] artifact_path = init_artifact(run_id) pipeline = build_pipeline(llm) with tracer.start_as_current_span("pr_review_pipeline") as span: span.set_attribute("review.run_id", run_id) span.set_attribute("review.diff_chars", len(diff)) initial_state: ReviewState = { "diff": diff, "artifact_path": artifact_path, "stage_results": [], "final_report": "", } final_state = pipeline.invoke(initial_state) span.set_attribute("review.stages_completed", len(final_state["stage_results"])) return final_state ``` > [!PULLQUOTE] > Clearing context between stages is only safe when state is durable. Without that durability, a multi-stage agent either stuffs every prior finding into every subsequent prompt or loses findings entirely. ## Verify it works Run the pipeline with the built-in mock LLM so no API key is needed. The block prints the final report, a per-stage latency table, and confirms the JSON artifact was written to disk. OTel spans are emitted to stdout via the `ConsoleSpanExporter`. ```python import sys sys.path.insert(0, "/workspace") import json from pathlib import Path from pipeline import run_review SAMPLE_DIFF = """ --- a/auth.py +++ b/auth.py @@ -10,7 +10,10 @@ def login(username, password): - query = f"SELECT * FROM users WHERE username='{username}'" + query = "SELECT * FROM users WHERE username=?" + cursor.execute(query, (username,)) token = jwt.encode({"sub": username}, SECRET_KEY) + if not token: + return None return token """ state = run_review(SAMPLE_DIFF, llm=None) # llm=None uses the mock print("\n=== FINAL REPORT ===") print(state["final_report"]) print("\n=== PER-STAGE LATENCY ===") for r in state["stage_results"]: print(f" {r['stage']:20s} {r['latency_ms']:7.1f} ms " f"tokens in={r['token_usage']['input_tokens']} " f"out={r['token_usage']['output_tokens']}") artifact = json.loads(Path(state["artifact_path"]).read_text()) stages_written = list(artifact["stages"].keys()) print(f"\nArtifact stages on disk: {stages_written}") assert len(stages_written) == 4, f"Expected 4 stages, got {len(stages_written)}" print("verify_pipeline_ok") ``` ## Step 6: Swap in a real LLM (requires API key) To run against Anthropic Claude, replace the `llm=None` argument with a real `ChatAnthropic` instance. The pipeline and tracing code are unchanged. ```python # Requires ANTHROPIC_API_KEY in environment import sys sys.path.insert(0, "/workspace") from langchain_anthropic import ChatAnthropic from pipeline import run_review llm = ChatAnthropic(model="claude-3-5-haiku-20241022", max_tokens=1024) SAMPLE_DIFF = """ --- a/auth.py +++ b/auth.py @@ -10,7 +10,10 @@ def login(username, password): - query = f"SELECT * FROM users WHERE username='{username}'" + query = "SELECT * FROM users WHERE username=?" + cursor.execute(query, (username,)) token = jwt.encode({"sub": username}, SECRET_KEY) + if not token: + return None return token """ state = run_review(SAMPLE_DIFF, llm=llm) print(state["final_report"]) ``` ## Reading the OTel spans The `ConsoleSpanExporter` writes each span as a JSON-like block to stdout. Each span includes the attributes you set in the stage functions. A `stage.security` span looks like this: ``` { "name": "stage.security", "context": {"trace_id": "0x...", "span_id": "0x..."}, "attributes": { "review.stage": "security", "llm.input_tokens": 312, "llm.output_tokens": 89, "stage.latency_ms": 1847.3 }, "parent_id": "0x..." <- links back to pr_review_pipeline root span } ``` The `parent_id` field links each stage span back to the root `pr_review_pipeline` span, giving you a complete trace tree. The same span structure indexes the same way on Datadog or Honeycomb. Only the exporter endpoint changes. For a persistent local backend, replace `ConsoleSpanExporter` with an OTLP exporter pointed at a local OpenTelemetry Collector writing to Grafana Tempo. The `tracing.py` module is the only file that changes. ## Troubleshooting **`ModuleNotFoundError: No module named 'langgraph'`** — The install block must run before any import block. If you're running files directly with `python`, activate the same environment where `uv pip install` ran, or prefix with `uv run python`. **`KeyError: 'stage_results'` on the first node** — The `Annotated[list, operator.add]` reducer requires the key to exist in the initial state dict. Confirm `initial_state` in `run_review` includes `"stage_results": []`. **Spans appear after the script exits, not inline** — You're using `BatchSpanProcessor`. Switch to `SimpleSpanProcessor(ConsoleSpanExporter())` in `tracing.py` for synchronous, inline output. `BatchSpanProcessor` is correct for production throughput but buffers until process exit. **Artifact file is empty or missing stages** — Check that `ARTIFACT_DIR` (`/tmp/review_artifacts`) is writable. On macOS with SIP restrictions, change `ARTIFACT_DIR` in `artifact_store.py` to a path inside your home directory. **`ChatAnthropic` raises `AuthenticationError`** — Verify `ANTHROPIC_API_KEY` is exported in the same shell session. `echo $ANTHROPIC_API_KEY` should print the key. The variable is not inherited by subprocesses started from a different terminal tab. **Token usage shows zeros with real LLM** — Some model versions return usage metadata under different keys. Print `response.usage_metadata` raw and adjust the `.get()` key names in `_call_llm` to match. ## Next steps - **Parallel security and logic stages** — Use LangGraph's `Send` API to fan out `stage_security` and `stage_logic` concurrently, then join before `stage_synthesise`. This cuts wall-clock time roughly in half for the two independent analysis stages. - **Persistent run history** — Replace the flat JSON artifact with a SQLite-backed store (using `sqlite3` from the standard library) so you can query findings across multiple PRs and track which issue types recur most often. - **OTLP export to Grafana Tempo** — Swap `ConsoleSpanExporter` for `opentelemetry-exporter-otlp-proto-grpc` pointed at a local Tempo instance. The span attributes defined here (`llm.input_tokens`, `stage.latency_ms`) become Tempo search tags immediately. - **Validation pass** — Add a fifth node that re-reads the artifact, checks whether any `CRITICAL` findings were introduced by the diff itself (not pre-existing), and sets a `review.passed` boolean span attribute. This mirrors the regression-revert logic in adamsreview [1]. ## FAQ ### Why store findings in JSON artifacts between pipeline stages? Persisting findings as JSON artifacts allows each stage to clear its context window without losing prior results, reducing token usage and noise. Without durable state, a multi-stage pipeline either repeats all prior findings in every prompt (expensive) or loses them entirely. ### How does the pipeline instrument latency and token usage per stage? Each stage function wraps its LLM call in an OpenTelemetry span, recording input and output token counts and wall-clock latency as span attributes. The ConsoleSpanExporter writes these spans to stdout as JSON, making per-stage performance visible without external services. ### What does the ConsoleSpanExporter output look like? Each span is a JSON-like block containing the span name, trace and span IDs, custom attributes (review.stage, llm.input_tokens, stage.latency_ms), and a parent_id linking it back to the root pr_review_pipeline span. This creates a complete trace tree in stdout. ### Can the pipeline run without an API key? Yes. The tutorial includes a mock LLM path that returns placeholder responses, allowing the full pipeline to execute and verify artifact persistence and tracing without credentials. Real LLM calls require ANTHROPIC_API_KEY or OPENAI_API_KEY in the environment. ### How would you export spans to a persistent backend instead of console? Replace ConsoleSpanExporter in tracing.py with an OTLP exporter (e.g., opentelemetry-exporter-otlp-proto-grpc) pointed at a local OpenTelemetry Collector or Grafana Tempo. The span attributes defined in the stages become searchable tags in the backend. ## References 1. https://github.com/adamjgmiller/adamsreview