plott-sdk-langgraph 0.0.1a0

Plott Analytics SDK for LangGraph - Automatic analytics tracking for LangGraph

Plott SDK for LangGraph

Automatic analytics tracking for LangGraph applications. Just wrap your graph and analytics are collected transparently.

Installation

pip install plott-sdk-langgraph

Quick Start

from langgraph.graph import StateGraph
from langchain_core.messages import HumanMessage
from plott_langgraph import PlottTrackedGraph

# Create your graph normally
workflow = StateGraph(AgentState)
workflow.add_node("chat", chat_node)
workflow.add_node("tools", tool_node)
workflow.add_edge("chat", "tools")
# ... add more nodes and edges

graph = workflow.compile(checkpointer=memory)

# Wrap with Plott - that's it!
tracked_graph = PlottTrackedGraph(graph, {
    "api_key": "cpk_...",  # or set PLOTT_API_KEY env var
    "environment": "development",
})

# Use exactly like the original graph - analytics are automatic
async for event in tracked_graph.astream_events(
    {"messages": [HumanMessage(content="Hello!")]},
    config={"configurable": {"thread_id": "123", "run_id": "run-456"}},
):
    # Your normal event handling
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)

What Gets Tracked

The SDK automatically tracks:

Event	Description
Messages	User input and assistant responses
Tool Calls	Tool executions with arguments and results
State Snapshots	State changes as the graph executes
Run Lifecycle	Start and end of each graph run
Errors	Any errors that occur during execution

How It Works

The SDK wraps your compiled LangGraph graph and intercepts astream_events():

1. Zero code changes - Your graph logic stays exactly the same
2. Non-blocking - Analytics are tracked in the background
3. Error-isolated - Analytics errors never affect your graph execution
4. Transparent - Events pass through unchanged

Your Code → TrackedGraph → Original Graph
                ↓
          EventTracker → PlottAnalytics → API

Configuration

tracked_graph = PlottTrackedGraph(graph, {
    "api_key": "cpk_...",           # Required (or PLOTT_API_KEY env var)
    "base_url": "https://...",       # Default: https://api.askwise.com
    "environment": "production",     # production, staging, development, test
    "batch_size": 10,                # Events per batch
    "flush_interval": 1.0,           # Seconds between auto-flushes
    "debug": False,                  # Enable debug logging
})

Environment Variables

• PLOTT_API_KEY - API key (if not provided in config)
• ASKWISE_API_PATH - Custom API base URL

Graceful Shutdown

To ensure all events are flushed before your application exits:

# When done
await tracked_graph.shutdown()

Using with Other Methods

The tracked graph proxies all methods to the original graph:

# These work exactly as before
result = await tracked_graph.ainvoke(input)
async for chunk in tracked_graph.astream(input):
    ...

# Get state
state = await tracked_graph.aget_state(config)

Only astream_events() has analytics tracking. Other methods pass through directly.

License

MIT