Skip to main content
GitHub

LangGraph

Auto-instrumentation for LangGraph agents.

Risicare provides deep integration with LangGraph for graph-based agent observability.

Version Compatibility

Requires langgraph >= 0.2.0 and langchain-core >= 0.2.0.

Installation

pip install risicare[langgraph]
# or
pip install risicare langgraph

Basic Usage

import risicare
from langgraph.graph import StateGraph, END
 
risicare.init()
 
# Define your graph as usual - it's automatically traced
def agent_node(state):
    return {"messages": state["messages"] + ["Agent response"]}
 
graph = StateGraph(dict)
graph.add_node("agent", agent_node)
graph.set_entry_point("agent")
graph.add_edge("agent", END)
 
app = graph.compile()
result = app.invoke({"messages": ["Hello"]})

What's Captured

Graph Structure

  • Node names and types
  • Edge connections
  • Entry and exit points

Execution Flow

  • Node invocations with timing
  • Graph name and thread ID
  • State key count at each step
  • Loop iterations

Agent Identity

Annotate nodes with agent identity:

from risicare import agent_context
 
def researcher_node(state):
    with agent_context("researcher-001", agent_name="researcher", agent_role="specialist"):
        # Research logic
        return {"findings": research_results}
 
def writer_node(state):
    with agent_context("writer-001", agent_name="writer", agent_role="specialist"):
        # Writing logic
        return {"draft": written_content}

Decision Phases

Track Think/Decide/Act within nodes:

from risicare import SemanticPhase, phase_context
 
def planning_node(state):
    with phase_context(SemanticPhase.THINK):
        analysis = analyze_task(state["task"])
 
    with phase_context(SemanticPhase.DECIDE):
        plan = create_plan(analysis)
 
    return {"plan": plan}
 
def execution_node(state):
    with phase_context(SemanticPhase.ACT):
        result = execute_plan(state["plan"])
 
    return {"result": result}

Conditional Routing

Conditional edges are visible through the span hierarchy:

def should_continue(state):
    if state["iteration"] >= 3:
        return "end"
    return "continue"
 
graph.add_conditional_edges(
    "agent",
    should_continue,
    {"continue": "agent", "end": END}
)

Routing decisions are reflected in which node spans execute next. The span tree shows the path taken through the graph, so you can infer routing outcomes from the execution sequence.

Subgraphs

Nested subgraphs maintain trace hierarchy:

# Parent graph
parent = StateGraph(dict)
parent.add_node("child_graph", child_app)  # Subgraph as node
 
# Child traces are nested under parent

Provider Spans

LangGraph instrumentation creates agent/framework-level spans. Underlying LLM calls (e.g., OpenAI, Anthropic) are traced separately by provider instrumentation, giving you both framework-level and LLM-level visibility. Note that LangChain callbacks handle deduplication when LangChain is also instrumented.

Visualization

View graph execution in the dashboard:

  • Timeline View: Node execution waterfall
  • Graph View: Visual graph with execution path highlighted

Next Steps