LangGraph - reagent-flow

Installation

uv add reagent-flow-langgraph

Setup

Use ReagentGraphTracer as a callback for your graph:

from reagent_flow_langgraph import ReagentGraphTracer
import reagent_flow

tracer = ReagentGraphTracer()

with reagent_flow.session("graph") as s:
    graph.invoke({"input": "..."}, config={"callbacks": [tracer]})

s.assert_called("my_tool")

How it works

ReagentGraphTracer extends the LangChain callback handler with graph-specific node tracking. It captures the same events as the LangChain handler, plus awareness of which graph node is currently executing.

Example with a ReAct agent

import json
from langgraph.prebuilt import create_react_agent
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.tools import tool
from reagent_flow_langgraph import ReagentGraphTracer
import reagent_flow

@tool
def extract_vendor_packet(vendor_name: str) -> str:
    """Return the intake packet for a vendor review."""
    return json.dumps({"vendor_name": vendor_name, "status": "captured"})

llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash")
agent = create_react_agent(llm, [extract_vendor_packet])
tracer = ReagentGraphTracer()

with reagent_flow.session("intake") as s:
    agent.invoke(
        {"messages": [("user", "Prepare a vendor packet for ClearVoice AI")]},
        config={"callbacks": [tracer]},
    )

s.assert_called("extract_vendor_packet")
s.assert_tool_output_matches("extract_vendor_packet", schema={
    "vendor_name": str,
    "status": str,
})

Multi-agent pipeline

LangGraph excels at multi-agent architectures. Use separate reagent-flow sessions for each sub-agent, linked via parent_trace_id and handoff_context:

# Agent A
with reagent_flow.session("intake", trace_dir=trace_dir) as a:
    agent_a.invoke(task_a, config={"callbacks": [ReagentGraphTracer()]})

# Extract structured output from agent A
vendor_packet = a.trace.turns[-1].tool_results[0].result

# Agent B receives the handoff
with reagent_flow.session(
    "security-review",
    trace_dir=trace_dir,
    parent_trace_id=a.trace.trace_id,
    handoff_context=vendor_packet,
) as b:
    agent_b.invoke(task_b, config={"callbacks": [ReagentGraphTracer()]})

# Contract at the boundary
b.assert_handoff_received(a)
b.assert_handoff_matches(schema={"vendor_name": str, "status": str})

See Vendor Onboarding Showcase for a complete multi-agent pipeline.