graphobs 0.2.1

Contract-first observability helpers for graph-based Python applications.

graphobs

Contract-first observability for graph-based Python apps.

graphobs helps teams describe graph state boundaries once, then reuse those declarations for trace payloads, structured logs, and validation.

The 0.2.1 release is intentionally small. It includes the core contract model, LangGraph integration helpers, callback payload projection, backend-portable tracing helpers, and structured logging helpers.

Why This Exists

Graph applications often grow faster than their observability model. Node inputs, node outputs, trace payloads, and log events can drift into separate conventions, which makes debugging harder and increases the chance of recording more state than intended.

This project starts from one design rule: state contracts, trace contracts, and log payload boundaries should be designed together.

Why This Instead Of Auto-Instrumentation?

Use auto-instrumentation first when you only need a zero-code execution trace. It is the lowest-friction baseline.

graphobs is for the next problem: deciding what each graph boundary is allowed to expose, then using that same contract to validate writes and curate OpenTelemetry/OpenInference payloads. Auto-instrumentors can show what happened, but they cannot infer which state keys are public, which are local scratch state, or whether a node wrote a key it should not own.

If your graph has fewer than three nodes or no noisy shared state, this library may be more structure than you need. If traces are full of whole-state dumps, private scratch values, or inconsistent node payloads, start with callback payload projection or one contract and migrate outward.

Why Not Agent Contracts Or PII Middleware?

agent-contracts explores declarative contracts for LangGraph agents, including graph construction and runtime contract enforcement. graphobs is narrower: keep your existing LangGraph shape, then use contracts to drive curated trace payloads, span attributes, structured logs, and write validation.

LangChain PII middleware is useful when you need to detect, redact, mask, or block PII in agent messages and tool outputs. graphobs is not a PII detector. It reduces overcollection by projecting only contract-declared graph state into telemetry and by using compact payload summaries by default.

Mental Model

• State is the working data a graph reads and writes.
• Logs record lifecycle events and correlation fields.
• Traces show execution flow, timing, inputs, outputs, attributes, and errors.
• Contracts define which parts of state are public at a graph boundary and which parts should stay local to an implementation detail.

Quickstart

Interactive notebook (recommended for new users)

Install the demo bundle and open the notebook:

pip install "graphobs[demo]"
jupyter lab examples/notebooks/quickstart.ipynb

The notebook walks through the full idea in two acts — no credentials required for Act 1 (curated vs messy span contrast in an embedded viewer), and a per-platform .env recipe for Act 2 (Arize Phoenix, LangSmith, MLflow, Langfuse).

Library install from a checkout

Install the project in editable mode from a checkout:

uv sync --all-groups
uv run python -c "import graphobs; print(graphobs.__version__)"

Run the local checks:

uv run ruff check .
uv run ruff format --check .
uv run mypy .
uv run pytest
uv run mkdocs build --strict
uv build

Examples

The examples/ directory contains local-only synthetic examples that compare raw LangGraph flows with contract-wrapped versions:

uv run python -m examples.simple_rag.app
uv run python -m examples.subgraph_boundary.app
uv run python -m examples.tool_agent.app
uv run python -m examples.backend_export.app

Each example prints deterministic JSON with raw spans, compact contract spans, an intentional contract validation error, and lifecycle log summaries.

Current Package Surface

The package root exposes the headline adoption path. Lower-level projection, logging, and tracing primitives remain available from focused submodules.

from graphobs import NodeContract
from graphobs.contracts.projection import project_input

contract = NodeContract(
    name="classify",
    reads=("request.text",),
    writes=("classification.label",),
)

state = {"request": {"text": "hello"}, "scratch": {"notes": "local"}}
public_input = project_input(contract, state)

The contract model is plain Python and has no graph runtime dependency.

For an existing LangGraph app, choose the adoption path by risk:

• Want cleaner callback payloads without changing execution? Start with project_callback_payloads.
• Want enforcement while preserving current node behavior? Use contract_node(..., pass_through_state=True, audit_reads=True, on_violation=ContractViolationAction.WARN).
• Want strict contract-shaped execution for a stable node? Use the default contract_node wrapper.

Callback projection is the lowest-risk migration path because the node still receives the graph state LangGraph would normally provide:

from graphobs.langgraph.callbacks import project_callback_payloads

config = {
    "callbacks": [
        project_callback_payloads(callback, [classify_contract], diagnostics=True),
    ],
}

The strict LangGraph wrapper projects the node's execution input before the node runs:

from graphobs import NodeContract, contract_node

@contract_node(
    NodeContract(
        name="classify",
        reads=("request.text",),
        writes=("classification.label",),
    )
)
def classify(state):
    return {"classification": {"label": "question"}}

For migration guardrails without execution filtering, wrap at registration time with pass-through execution and read auditing:

from graphobs import NodeContract, contract_node
from graphobs.contracts.models import ContractViolationAction

classify_contract = NodeContract(
    name="classify",
    reads=("request.text",),
    writes=("classification.label",),
)

graph.add_node(
    "classify",
    contract_node(
        classify,
        classify_contract,
        pass_through_state=True,
        audit_reads=True,
        on_violation=ContractViolationAction.WARN,
    ),
)

Tracing helpers emit OpenTelemetry spans with OpenInference semantic attributes. Exporter configuration stays outside the library, so applications can choose any OpenTelemetry-compatible backend.

from graphobs.tracing import start_graph_span

with start_graph_span(
    "classify",
    "CHAIN",
    input=public_input,
    attributes={"graph.node": "classify"},
):
    ...

Payloads use compact structural summaries by default. Explicit full payload mode is available for controlled debugging, but it is unsafe for sensitive production data.

Structured logging helpers emit lifecycle events with correlation fields and durations. They do not configure a logging backend or store full graph state.

from graphobs import build_invoke_config
from graphobs.logging.context import LogContext

config = build_invoke_config(
    LogContext(session_id="session-1", request_id="request-1"),
)
graph.compile().invoke({"request": {"text": "hello"}}, config=config)

Public Neutrality

All docs, tests, examples, fixtures, comments, and exported APIs must use synthetic, generic concepts. Do not include organization-specific, product-specific, deployment-specific, or private runtime details.

See docs/concepts/public-neutrality.md for the repository policy.

Documentation

• Quickstart Notebook
• Architecture
• Examples
• State, Logs, And Traces
• Public Vs Private Graph State
• Designing Node Contracts
• Payload Safety And Redaction
• Backend Portability With OTel And OpenInference
• Migrate One Node At A Time
• Medium Companion
• 0.2.1 Release Notes
• Contracts API Reference
• LangGraph API Reference
• Logging API Reference
• Tracing API Reference
• Release And Versioning Notes

The MkDocs site can be built locally with:

uv run mkdocs build --strict