LangGraph

pip install netra-sdk langgraph

from netra import Netra
from langgraph.graph import StateGraph
import os

# Initialize Netra
Netra.init(
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}",
    trace_content=True
)

# Define your graph - automatically traced
from typing import TypedDict

class GraphState(TypedDict):
    messages: list[str]

workflow = StateGraph(GraphState)

from netra.decorators import workflow, agent, task
from netra import SpanWrapper

# Node function with task decorator
@task()
def process_node(state: GraphState) -> GraphState:
    span = SpanWrapper("node-processing", {
        "node.name": "process",
        "state.messages": len(state["messages"])
    }).start()
    
    result = {
        "messages": state["messages"] + ["Processed"]
    }
    
    span.end()
    return result

# Build graph with workflow decorator
@workflow()
def build_graph():
    workflow.add_node("process", process_node)
    workflow.set_entry_point("process")
    workflow.set_finish_point("process")
    
    return workflow.compile()

@agent()
def agent_workflow(query: str):
    graph = StateGraph(GraphState)
    
    @task()
    def analyze(state: GraphState) -> GraphState:
        return {"messages": state["messages"] + ["Analyzed"]}
    
    @task()
    def decide(state: GraphState) -> GraphState:
        return {"messages": state["messages"] + ["Decision made"]}
    
    graph.add_node("analyze", analyze)
    graph.add_node("decide", decide)
    graph.add_edge("analyze", "decide")
    graph.set_entry_point("analyze")
    graph.set_finish_point("decide")
    
    app = graph.compile()
    return app.invoke({"messages": [query]})

from netra import SpanWrapper
import json

def run_with_state_tracking(app, initial_state: GraphState):
    state_span = SpanWrapper("state-management").start()
    
    try:
        result = app.invoke(initial_state)
        state_span.set_attribute("state.initial", json.dumps(initial_state))
        state_span.set_attribute("state.final", json.dumps(result))
        state_span.end()
        return result
    except Exception as e:
        state_span.set_status(code=1, message=str(e))
        state_span.end()
        raise