Observability

NeoGraph provides observability at two layers. The framework layer (structlog) is always on and captures every node execution, LLM call, tool invocation, and budget event. The tracing layer (Langfuse or any LangChain callback) is opt-in and provides deep LLM call traces with token-level detail.

NeoGraph does not own the observability backend. It emits structured logs and threads callback configuration through the LangGraph runtime. You choose where the data goes.

structlog: always on

Every NeoGraph operation emits structured log events via structlog. This happens automatically — no configuration required.

What gets logged

Event	Fields	When
`compile_start`	`construct`, `nodes`, `node_names`, `modifiers`	`compile()` is called
`compile_complete`	`construct`, `state_fields`	Graph compilation finishes
`subgraph_compile`	`subgraph`, `input`, `output`	Sub-Construct compiled
`node_start`	`node`, `mode`, `model`, `prompt`, `tools`, `budgets`	Node execution begins
`node_complete`	`node`, `mode`, `duration_s`	Node execution finishes
`llm_call`	`tier`, `prompt`, `mode`, `duration_s`, `input_tokens`, `output_tokens`, `total_tokens`	LLM call completes
`tool_call`	`tool`, `call_num`, `duration_s`	Individual tool invocation
`tool_budget_exhausted`	`tool`	Tool hits its budget cap
`all_tools_exhausted`	`exhausted`, `forcing_response`	All budgeted tools spent
`react_final_response`	`loop`	ReAct loop produces final answer

Example output

A think node emits something like this (formatted as JSON for clarity):

{
  "node": "classify",
  "mode": "think",
  "model": "reason",
  "prompt": "rw/classify",
  "event": "node_start",
  "output_type": "ClassifiedClaims"
}
{
  "tier": "reason",
  "prompt": "rw/classify",
  "mode": "think",
  "duration_s": 1.847,
  "input_tokens": 2340,
  "output_tokens": 512,
  "total_tokens": 2852,
  "event": "llm_call"
}
{
  "node": "classify",
  "mode": "think",
  "duration_s": 1.851,
  "event": "node_complete"
}

An agent node with tools adds tool-level detail:

{
  "node": "explore",
  "mode": "agent",
  "model": "reason",
  "prompt": "rw/explore",
  "tools": ["search_nodes", "read_artifact"],
  "budgets": {"search_nodes": 5, "read_artifact": 10},
  "event": "node_start"
}
{
  "tool": "search_nodes",
  "call_num": 1,
  "duration_s": 0.234,
  "event": "tool_call"
}
{
  "tool": "search_nodes",
  "event": "tool_budget_exhausted"
}
{
  "tier": "reason",
  "prompt": "rw/explore",
  "mode": "react",
  "loops": 4,
  "tool_calls": 7,
  "duration_s": 8.312,
  "input_tokens": 12400,
  "output_tokens": 1890,
  "total_tokens": 14290,
  "output": "ResearchFindings",
  "event": "llm_call"
}

Configuring structlog

NeoGraph calls structlog.get_logger() — it does not configure structlog itself. You configure structlog in your application the way you normally would:

import structlog

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.dev.ConsoleRenderer(),  # pretty for dev
        # or structlog.processors.JSONRenderer() for prod
    ],
)

If you do not configure structlog, it uses its defaults (which are reasonable for development).

Injecting trace providers via FromConfig

Use Annotated[T, FromConfig] to inject trace providers, span managers, or any shared observability resource into scripted nodes without threading them through state:

from typing import Annotated
from neograph import node, FromConfig

@node(outputs=Report)
def summarize(
    claims: Claims,
    tracer: Annotated[LangfuseTracer, FromConfig],
) -> Report:
    with tracer.span("summarize-claims"):
        result = Report(summary=f"{len(claims.items)} claims")
    return result

At runtime, tracer is resolved from config["configurable"]["tracer"]. Pass it when calling run():

from langfuse import Langfuse

langfuse = Langfuse()
tracer = langfuse.trace(name="ingestion-pipeline")

result = run(
    graph,
    input={"node_id": "BR-042"},
    config={"configurable": {"tracer": tracer}},
)

This pattern works for any shared resource — rate limiters, database pools, metrics collectors — not just tracing. The key is that Annotated[T, FromConfig] reads from config["configurable"], which is propagated to every node by the LangGraph runtime.

The config[“configurable”] pattern

All shared resources and runtime metadata flow through config["configurable"]. When you call run(), input fields are automatically injected there too:

result = run(
    graph,
    input={"node_id": "BR-042"},
    config={
        "configurable": {
            "tracer": langfuse_tracer,
            "rate_limiter": my_limiter,
        }
    },
)

# Inside every node, config["configurable"] contains:
# {
#     "node_id": "BR-042",        # from input
#     "tracer": <LangfuseTracer>, # from config
#     "rate_limiter": <Limiter>,  # from config
# }

This means your prompt compiler, LLM factory, and scripted functions can all access pipeline metadata and shared resources uniformly.

Langfuse: opt-in LLM tracing

Langfuse provides deep LLM call tracing — prompt/completion pairs, token counts, latency waterfalls, cost tracking. NeoGraph wires it for you through observe=.

One-line setup: `observe=True`

from neograph import compile, run

graph = compile(pipeline)

# Auto-attaches the Langfuse CallbackHandler and flushes on completion.
result = run(graph, input={"node_id": "BR-042"}, observe=True)

observe=True (equivalently observe="langfuse") is a per-run option on every verb — run, arun, stream, astream. It:

gates on the env keys — attaches only when both LANGFUSE_SECRET_KEY and LANGFUSE_PUBLIC_KEY are set, so it is a clean no-op offline and in CI (never crashes when keys are absent);
merges, never clobbers — if you already pass handlers in config["callbacks"], the Langfuse handler is added alongside them (and it will not double-attach if you already wired a Langfuse handler yourself);
flushes on completion — for run/arun after the call, and for stream/astream after the generator is exhausted (or closed early), so no trace batch is stranded.

Install the extra to enable it: pip install "neograph[langfuse]".

Every LLM call inside every node — think, agent, act, and even Oracle merge calls — appears in your Langfuse dashboard with full prompt/completion detail, token counts, and timing.

Escape hatch: attach the handler yourself

observe= covers the 90% case. When you need full control — a custom handler, extra Langfuse config, your own flush timing — attach the handler through the standard config["callbacks"] mechanism and flush manually:

from langfuse.langchain import CallbackHandler
from langfuse import get_client
from neograph import compile, run

graph = compile(pipeline)
result = run(
    graph,
    input={"node_id": "BR-042"},
    config={"callbacks": [CallbackHandler()]},
)
get_client().flush()

observe= merges into config["callbacks"], so the two forms compose — but pick one Langfuse handler per run to avoid duplicate traces.

How it works

NeoGraph does not integrate with Langfuse directly. The mechanism is LangGraph’s standard callback threading:

You pass callbacks=[langfuse_handler] in the config dict to run().
LangGraph propagates that config to every node invocation.
NeoGraph’s LLM layer calls llm.invoke(messages, config=config), which passes the callbacks to the underlying LangChain model.
The LangChain model fires callback events that Langfuse (or any other handler) captures.

This means any LangChain-compatible callback works. If you use LangSmith, Weights & Biases, or a custom handler, pass it the same way:

config = {"callbacks": [my_langsmith_handler, my_custom_handler]}
result = run(graph, input={...}, config=config)

Combining both layers

The two layers complement each other:

structlog gives you the framework-level view: which node ran, how long it took, how many tool calls, total tokens. Useful for pipeline debugging and operational monitoring.
Langfuse/callbacks give you the LLM-level view: exact prompts, completions, per-call token breakdown, cost. Useful for prompt engineering and model evaluation.

Both run simultaneously. structlog is always on; Langfuse activates only when you pass observe=True (or attach a handler yourself).