LangGraph

pip install netra-sdk langgraph

npm 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)

import { Netra } from "netra-sdk";
import { StateGraph } from "@langchain/langgraph";

async function main() {
  // Initialize Netra (must await)
  await Netra.init({
    headers: `x-api-key=${process.env.NETRA_API_KEY}`,
    traceContent: true
  });

  // Define your graph - automatically traced
  interface GraphState {
    messages: string[];
  }

  const workflow = new StateGraph<GraphState>({
    channels: {
      messages: { value: (x, y) => x.concat(y) }
    }
  });
}

main();

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()

import { workflow, agent, task, SpanWrapper } from "netra-sdk";

// Node function with task decorator
@task()
async function processNode(state: GraphState) {
  const span = new SpanWrapper("node-processing", {
    "node.name": "process",
    "state.messages": state.messages.length
  }).start();
  
  const result = {
    messages: [...state.messages, "Processed"]
  };
  
  span.end();
  return result;
}

// Build graph with workflow decorator
@workflow()
async function buildGraph() {
  workflow.addNode("process", processNode);
  workflow.setEntryPoint("process");
  workflow.setFinishPoint("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]})

@agent()
async function agentWorkflow(query: string) {
  const graph = new StateGraph<GraphState>({
    channels: {
      messages: { value: (x, y) => x.concat(y) }
    }
  });
  
  @task()
  async function analyze(state: GraphState) {
    return { messages: [...state.messages, "Analyzed"] };
  }
  
  @task()
  async function decide(state: GraphState) {
    return { messages: [...state.messages, "Decision made"] };
  }
  
  graph.addNode("analyze", analyze);
  graph.addNode("decide", decide);
  graph.addEdge("analyze", "decide");
  graph.setEntryPoint("analyze");
  graph.setFinishPoint("decide");
  
  const app = graph.compile();
  return await 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

import { SpanWrapper } from "netra-sdk";

async function runWithStateTracking(app: any, initialState: GraphState) {
  const stateSpan = new SpanWrapper("state-management").start();
  
  try {
    const result = await app.invoke(initialState);
    stateSpan.setAttribute("state.initial", JSON.stringify(initialState));
    stateSpan.setAttribute("state.final", JSON.stringify(result));
    stateSpan.end();
    return result;
  } catch (error) {
    stateSpan.setStatus({ code: 1, message: String(error) });
    stateSpan.end();
    throw error;
  }
}