Graphite

Graphite is an event-driven Python framework for building domain-specific AI agents, published as the grafi package. It composes assistants from workflows, nodes, tools, and topics, and instruments each of those layers with OpenTelemetry. Graphite reads its tracer from the ExecutionServices bundle you hand to GrafiRuntime, so pointing it at Arize AX is a matter of giving it a tracer from an Arize AX tracer provider. The framework’s own spans then nest assistant → workflow → node → tool, and the OpenInference OpenAI instrumentor adds the LLM span underneath with prompts, token counts, and cost.

Do not use Graphite’s own setup_tracing() helper for Arize AX. It builds a plaintext gRPC exporter (insecure=True) with no request headers, so exports to Arize AX fail with StatusCode.UNAVAILABLE ... Connection reset by peer, and the resource it builds carries service.name instead of the openinference.project.name that Arize AX reads to assign a project. Build the provider with arize.otel.register instead, as below.

Prerequisites

Python 3.11+ (required by grafi)
An Arize AX account (sign up)
Your Arize AX Space ID and API Key
An OPENAI_API_KEY from the OpenAI Platform

Launch Arize AX

Sign in to your Arize AX account.
From Space Settings, copy your Space ID and API Key. You will set them as ARIZE_SPACE_ID and ARIZE_API_KEY below.

Install

grafi already depends on openinference-instrumentation-openai and the OpenTelemetry SDK, so only the Arize AX exporter is extra:

pip install grafi arize-otel

Configure credentials

export ARIZE_SPACE_ID="<your-space-id>"
export ARIZE_API_KEY="<your-api-key>"
export ARIZE_PROJECT_NAME="graphite-tracing-example"
export OPENAI_API_KEY="<your-openai-api-key>"

Setup tracing

register returns the tracer provider. Instrument OpenAI with it for the LLM spans, then hand Graphite a tracer taken from it. Graphite names its span attributes after its own metadata model, so the span kind arrives as oi_span_type, the payloads as input and output, and the conversation as conversation_id. Arize AX reads the OpenInference names, so without a rename Graphite’s spans show with no span kind and no session. Graphite calls set_attribute on whatever tracer you give it, so wrapping the tracer is enough — the wrapper hands back a span proxy that renames keys on the way in, and changes nothing else:

# instrumentation.py
import os
from contextlib import contextmanager
from typing import Any, Iterator, Mapping

from arize.otel import register
from openinference.instrumentation.openai import OpenAIInstrumentor
from openinference.semconv.trace import SpanAttributes

tracer_provider = register(
    space_id=os.environ["ARIZE_SPACE_ID"],
    api_key=os.environ["ARIZE_API_KEY"],
    project_name=os.environ["ARIZE_PROJECT_NAME"],
)

OpenAIInstrumentor().instrument(tracer_provider=tracer_provider)

_RENAME = {
    "oi_span_type": SpanAttributes.OPENINFERENCE_SPAN_KIND,
    "input": SpanAttributes.INPUT_VALUE,
    "output": SpanAttributes.OUTPUT_VALUE,
    "conversation_id": SpanAttributes.SESSION_ID,
    "user_id": SpanAttributes.USER_ID,
}


class _OISpan:
    """Span proxy that renames Graphite's attribute keys."""

    def __init__(self, span: Any) -> None:
        self._span = span

    def set_attribute(self, key: str, value: Any) -> None:
        # One Graphite metadata field is a dict, which OpenTelemetry
        # rejects with "Invalid type dict for attribute"; drop it.
        if isinstance(value, Mapping):
            return
        if isinstance(value, (list, tuple, set)) and not all(
            isinstance(v, (str, int, float, bool)) for v in value
        ):
            return
        if key in ("conversation_id", "user_id") and not value:
            return
        self._span.set_attribute(_RENAME.get(key, key), value)

    def set_attributes(self, attributes: Mapping[str, Any]) -> None:
        for key, value in attributes.items():
            self.set_attribute(key, value)

    def __getattr__(self, name: str) -> Any:
        return getattr(self._span, name)


class OpenInferenceTracer:
    """Wraps a tracer so Graphite's spans use OpenInference attributes."""

    def __init__(self, tracer: Any) -> None:
        self._tracer = tracer

    @contextmanager
    def start_as_current_span(
        self, *args: Any, **kwargs: Any
    ) -> Iterator[_OISpan]:
        with self._tracer.start_as_current_span(*args, **kwargs) as span:
            yield _OISpan(span)

    def start_span(self, *args: Any, **kwargs: Any) -> _OISpan:
        return _OISpan(self._tracer.start_span(*args, **kwargs))

    def __getattr__(self, name: str) -> Any:
        return getattr(self._tracer, name)


tracer = OpenInferenceTracer(tracer_provider.get_tracer("grafi"))
print("Arize AX tracing initialized for Graphite.")

The rename passes Graphite’s own values through rather than second-guessing them, so each layer is labeled the way Graphite labels it — the workflow as an agent span, the node as a chain span, its OpenAI tool as an LLM span. Drop the wrapper and hand tracer_provider.get_tracer("grafi") straight to Graphite if you would rather see its raw attribute names.

For EU spaces, pass endpoint=Endpoint.ARIZE_EUROPE to register (import Endpoint from arize.otel).

Run Graphite

Pass the tracer in ExecutionServices. GrafiRuntime.invoke binds those services for the duration of the invocation, so every assistant, workflow, node, and tool span lands in the same trace:

# example.py

# Importing instrumentation first ensures the OpenAI instrumentor is
# installed before the assistant is built.
from instrumentation import tracer

import asyncio
import os
import uuid
from typing import Optional

from grafi.assistants.assistant import Assistant
from grafi.common.events.topic_events.publish_to_topic_event import (
    PublishToTopicEvent,
)
from grafi.common.models.invoke_context import InvokeContext
from grafi.common.models.message import Message
from grafi.nodes.node import Node
from grafi.runtime import ExecutionServices, GrafiRuntime
from grafi.tools.llms.impl.openai_tool import OpenAITool
from grafi.topics.topic_impl.input_topic import InputTopic
from grafi.topics.topic_impl.output_topic import OutputTopic
from grafi.workflows.impl.event_driven_workflow import EventDrivenWorkflow
from openinference.semconv.trace import OpenInferenceSpanKindValues
from pydantic import Field


class HaikuAssistant(Assistant):
    """One LLM node: input topic -> OpenAI -> output topic."""

    oi_span_type: OpenInferenceSpanKindValues = Field(
        default=OpenInferenceSpanKindValues.AGENT
    )
    name: str = Field(default="HaikuAssistant")
    type: str = Field(default="HaikuAssistant")
    api_key: Optional[str] = Field(
        default_factory=lambda: os.getenv("OPENAI_API_KEY")
    )
    model: str = Field(default="gpt-5.4-mini")

    def _construct_workflow(self) -> "HaikuAssistant":
        agent_input_topic = InputTopic(name="agent_input_topic")
        agent_output_topic = OutputTopic(name="agent_output_topic")

        llm_node = (
            Node.builder()
            .name("OpenAINode")
            .subscribe(agent_input_topic)
            .tool(
                OpenAITool.builder()
                .name("OpenAITool")
                .api_key(self.api_key)
                .model(self.model)
                .system_message("Reply with a single haiku.")
                .build()
            )
            .publish_to(agent_output_topic)
            .build()
        )

        self.workflow = (
            EventDrivenWorkflow.builder()
            .name("HaikuWorkflow")
            .node(llm_node)
            .build()
        )
        return self


async def main() -> None:
    runtime = GrafiRuntime(ExecutionServices(tracer=tracer))
    assistant = HaikuAssistant()

    event = PublishToTopicEvent(
        invoke_context=InvokeContext(
            conversation_id="graphite-tracing-demo",
            invoke_id=uuid.uuid4().hex,
            assistant_request_id=uuid.uuid4().hex,
        ),
        data=[Message(content="Write a haiku about tracing.", role="user")],
    )

    async for output in runtime.invoke(assistant, event, is_sequential=True):
        print(output.data[0].content)


if __name__ == "__main__":
    asyncio.run(main())

python example.py

Expected output

Arize AX tracing initialized for Graphite.
Lines softly reveal
Hidden paths wake under light
Tracing finds the way

Verify in Arize AX

Open your Arize AX space and select project graphite-tracing-example.
Open the newest trace. Graphite’s layers nest, with the OpenInference LLM span at the leaf:

HaikuAssistant.run        (agent span)
  HaikuWorkflow.invoke      (agent span)
    OpenAINode.invoke         (chain span)
      OpenAITool.invoke         (LLM span)
        ChatCompletion            (LLM span)

Check from the skill, CLI, or SDK

Confirm spans are actually reaching your Arize AX project. Use whichever fits your workflow — the skill and CLI work for any framework; the SDK check is shown for each language.

Arize skill (agent)
AX CLI
SDK

Install the Arize Skills plugin and let your coding agent check for you:

npx skills add Arize-ai/arize-skills

Then prompt your agent:

Use the arize-trace skill to export and analyze recent traces from my project. Confirm spans are arriving, and summarize any errors or latency issues.

Export recent spans for your project — any rows mean traces are landing:

ax spans export "$ARIZE_PROJECT_NAME" --space "$ARIZE_SPACE_ID" \
  --limit 5 --stdout | jq 'length'

A non-zero count confirms spans reached Arize AX. Run ax auth login first if you have not authenticated. See the ax spans reference.

Query the project’s spans and check that at least one came back.

import os
from arize import ArizeClient

client = ArizeClient(api_key=os.environ["ARIZE_API_KEY"])
resp = client.spans.list(
    project=os.environ["ARIZE_PROJECT_NAME"],
    space=os.environ["ARIZE_SPACE_ID"],
    limit=5,
)
count = len(resp.spans)
print(
    f"{count} span(s) found" if count else "No spans yet — recheck setup"
)

// Reads ARIZE_API_KEY from the environment.
import { listSpans } from "@arizeai/ax-client";

const { data: spans } = await listSpans({
  project: process.env.ARIZE_PROJECT_NAME!,
  space: process.env.ARIZE_SPACE_ID!,
  limit: 5,
});
const count = spans.length;
console.log(
  count ? `${count} span(s) found` : "No spans yet — recheck setup",
);

client, err := arize.NewClient(
    arize.Config{APIKey: os.Getenv("ARIZE_API_KEY")},
)
if err != nil {
    log.Fatal(err)
}
resp, err := client.Spans.List(ctx, spans.ListRequest{
    Project: os.Getenv("ARIZE_PROJECT_NAME"),
    Space:   os.Getenv("ARIZE_SPACE_ID"),
    Limit:   5,
})
if err != nil {
    log.Fatal(err)
}
fmt.Printf("%d span(s) found\n", len(resp.Spans))

SDK span references: Python · TypeScript · Go.

What Arize AX captures

One span per Graphite layer — assistant, workflow, node, and tool — carrying its name, type, latency_ms, and the invocation’s conversation_id, invoke_id, and assistant_request_id
A fully attributed LLM span from the OpenInference OpenAI instrumentor: prompts and completions, llm.model_name, token counts, and cost
The assistant’s input and output payloads, size-bounded by Graphite
The invocation’s conversation_id as the session, so a multi-turn conversation groups in the Sessions view

The span kinds, input.value / output.value, and the session come from the rename in Setup tracing. Without it Graphite’s own spans arrive under its own attribute names, so they show with no span kind and no session grouping. A rename in Graphite itself would make the wrapper unnecessary.

Troubleshooting

Invalid type dict for attribute 'kwargs' value on startup. Graphite copies its metadata model onto the span and one field is a dictionary, which OpenTelemetry rejects. The wrapper in Setup tracing drops that field, so the warning means the wrapper is not in the path — check that ExecutionServices is given the wrapped tracer.
StatusCode.UNAVAILABLE or Connection reset by peer when exporting. Something is exporting plaintext gRPC to Arize AX — usually Graphite’s setup_tracing(). Build the provider with arize.otel.register as shown in Setup tracing.
Traces land in a project you did not name. Only register sets openinference.project.name. If a Graphite helper created the provider instead, the resource carries service.name and Arize AX cannot map it to your project.
RuntimeError: No ExecutionServices bound for the current invocation. The assistant was invoked directly instead of through runtime.invoke(...). Either invoke through the runtime, or wrap the call in grafi.runtime.bind_services(...).
Graphite’s spans show with no span kind. ExecutionServices was given the raw tracer rather than the OpenInferenceTracer wrapper from Setup tracing.
Spans stop at the assistant with no LLM child. instrumentation must be imported before the assistant is constructed, so OpenAIInstrumentor is in place when the OpenAI client is created.

OpenTelemetry

LLM Providers

Agent Frameworks

Coding Agents

Platforms

Orchestration

Evaluation

Prerequisites

Launch Arize AX

Install

Configure credentials

Setup tracing

Run Graphite

Expected output

Verify in Arize AX

Check from the skill, CLI, or SDK

What Arize AX captures

Troubleshooting

Resources

Graphite documentation

Graphite on GitHub

OpenInference OpenAI Instrumentor

arize-otel configuration

​Prerequisites

​Launch Arize AX

​Install

​Configure credentials

​Setup tracing

​Run Graphite

​Expected output

​Verify in Arize AX

​Check from the skill, CLI, or SDK

​What Arize AX captures

​Troubleshooting

​Resources

Graphite documentation

Graphite on GitHub

OpenInference OpenAI Instrumentor

arize-otel configuration

Prerequisites

Launch Arize AX

Install

Configure credentials

Setup tracing

Run Graphite

Expected output

Verify in Arize AX

Check from the skill, CLI, or SDK

What Arize AX captures

Troubleshooting

Resources