Instrument and Verify a Voice Agent: Make Your Calls Appear, Then Prove It Worked

Install

Everything new lands in one directory. Your agent gains one import, one span opened in the right place, and one call as it ends.

your-repo/
├── observability/
│   ├── __init__.py              # empty. Both listings import by package path
│   └── futureagi/
│       ├── __init__.py          # empty
│       ├── setup.py             # provider, mapper, capture             V2
│       ├── fi_verify_voice.py   # the checker, downloaded below         all twelve
│       ├── voice_spans.py       # every voice attribute key, once       V6 V7 V8 V9 V10
│       └── livekit_pii_alias.py # one shim, only on LiveKit             V11
├── agent.py                     # + the conversation span, opened early V3 V4 V5
├── requirements.txt             # + fi-instrumentation-otel
└── .gitignore                   # + .fi_verify/

Every step below offers two tracks, the Future AGI SDK and plain OpenTelemetry. Pick one and stay on it from here to the end. The two tracks write files of the same name that are not interchangeable, so every listing names its own track on the first line.

# Future AGI SDK track
python3.11 -m venv .venv && source .venv/bin/activate   # every listing below says `python`
pip install fi-instrumentation-otel      # not fi-instrumentation. Python 3.11+
pip install traceai-livekit              # or traceai-pipecat, for the framework you run
pip install "livekit-agents[openai]"     # the framework itself. traceai-livekit does not
                                         # depend on it, so nothing else pulls it in

# OpenTelemetry track
python3.11 -m venv .venv && source .venv/bin/activate   # every listing below says `python`
pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-http

Then the checker. One file, pure standard library, nothing to install, and Step 1 runs it.

mkdir -p observability/futureagi
curl -fsSL https://docs.futureagi.com/fi_verify_voice.py -o observability/futureagi/fi_verify_voice.py
shasum -a 256 observability/futureagi/fi_verify_voice.py
# 9e487b4e8eb00c1adfb57e2cfdda182005cb8f99d85e909e6531d7730380be2f

touch observability/__init__.py observability/futureagi/__init__.py
echo ".fi_verify/" >> .gitignore

export FI_API_KEY="your-api-key"
export FI_SECRET_KEY="your-secret-key"
export FI_PROJECT_NAME="my-voice-agent"

What a verified integration means

Voice has one failure the text integration does not, and it is the one that costs you the whole product surface.

Your call can be a perfect trace and still not be a call. The Voice tab selects a span that is typed as a conversation, has no parent, sits in the project, is not deleted, and falls inside the window on both its event time and its arrival time. Miss the parent condition alone and the call is in Traces, correctly shaped, fully populated, and absent from the Voice tab, from every voice filter, and from every voice eval. Nothing errors, and no dashboard reads differently.

Two more that look like nothing:

• The Duration, Turns and Talk ratio columns read named attributes off that one span. Nothing derives them from the audio or from the child spans. Miss the name and the column is blank
• The transcript is read under three different keys by three different surfaces, and none of them falls back to another. Write two of the three and the call looks complete on one screen and empty on the next

So the result is decided by a checker rather than by looking. Twelve gates, each tied to the step that closes it:

Eleven gates read a local capture of your spans. V1 reads two receipts written during the run, because a capture says nothing about whether anything arrived.

Tutorial

python observability/futureagi/fi_verify_voice.py preflight

# observability/futureagi/setup.py     Future AGI SDK track
import os

from fi_instrumentation import FITracer, register
from fi_instrumentation.fi_types import ProjectType
from traceai_livekit import enable_http_attribute_mapping

from . import fi_verify_voice, livekit_pii_alias

# Import once at process start, AFTER whatever loads your .env, and before any
# LiveKit import that builds a session. Imported earlier the keys are not there
# yet, and only V1 says so.
provider = register(
    project_name=os.environ["FI_PROJECT_NAME"],
    project_type=ProjectType.OBSERVE,
    set_global_tracer_provider=True,      # LiveKit's own spans need the global provider
)

# 1. swap FI's exporter for the one that maps LiveKit attributes.
enable_http_attribute_mapping()

# 2. put the conversation content back on the keys that mapper reads.
livekit_pii_alias.install(provider)

# 3. capture what was really sent, at the exporter, after the mapping.
if os.getenv("FI_VERIFY", "1") == "1":
    fi_verify_voice.attach(provider)

# FITracer, not get_tracer(): a plain Tracer drops session.id and user.id, failing V5
tracer = FITracer(provider.get_tracer("voice-agent"))

# observability/futureagi/livekit_pii_alias.py     Future AGI SDK track
ALIAS = {
    "lk.pii.user_input": "lk.user_input",
    "lk.pii.chat_ctx": "lk.chat_ctx",
    "lk.pii.response.text": "lk.response.text",
    "lk.pii.response.function_calls": "lk.response.function_calls",
    "lk.pii.function_tool.arguments": "lk.function_tool.arguments",
    "lk.pii.function_tool.output": "lk.function_tool.output",
    "lk.pii.input_text": "lk.input_text",
    "lk.pii.instructions": "lk.instructions",
    "lk.pii.room_name": "lk.room_name",
    "lk.pii.user_transcript": "lk.user_transcript",
    "lk.pii.participant_identity": "lk.participant_identity",
}


def install(provider):
    """Wrap every exporter on the provider. Call after enable_http_attribute_mapping()."""
    active = getattr(provider, "_active_span_processor", None)
    procs = list(getattr(active, "_span_processors", ())) or ([active] if active else [])
    for proc in procs:
        exp = getattr(proc, "span_exporter", None) or getattr(proc, "_exporter", None)
        if exp is None or getattr(exp, "_lk_pii_alias", False):
            continue
        real = exp.export

        def export(spans, _real=real):
            for s in spans:
                a = getattr(s, "_attributes", None)
                if not a:
                    continue
                add = {new: a[old] for old, new in ALIAS.items() if old in a and new not in a}
                if add:
                    s._attributes = {**dict(a), **add}
            return _real(spans)

        exp.export = export
        exp._lk_pii_alias = True

# observability/futureagi/setup.py     OpenTelemetry track
import os, contextvars

from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider, SpanProcessor
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

from . import fi_verify_voice

# Import this AFTER whatever loads your .env, or the keys are not there yet and only V1 says so.
# On the RESOURCE: without it the collector answers 400 and the client still looks healthy
resource = Resource.create({"project_name": os.environ["FI_PROJECT_NAME"],
                            "project_type": "observe"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(
    endpoint="https://api.futureagi.com/tracer/v1/traces",
    headers={"X-Api-Key": os.environ["FI_API_KEY"],
             "X-Secret-Key": os.environ["FI_SECRET_KEY"]})))

# Step 4 sets this once at the edge; every span picks it up here, so no call site remembers it
_scope = contextvars.ContextVar("fi_scope", default={})


class FiScope(SpanProcessor):     # the base class no-ops the other three methods
    def on_start(self, span, parent_context=None):
        for k, v in _scope.get().items():
            span.set_attribute(k, v)


provider.add_span_processor(FiScope())
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("voice-agent")

if os.getenv("FI_VERIFY", "1") == "1":
    fi_verify_voice.attach(provider)

# agent.py     Future AGI SDK track
from observability.futureagi.setup import provider, tracer     # first import, before livekit
from fi_instrumentation import using_attributes
from livekit.agents import AgentSession

session = AgentSession(stt=..., llm=..., tts=...)

# The conversation span opens BEFORE session.start(). Opened after it, LiveKit's own
# agent_session span is already the root, this one becomes a child, and the Voice tab
# never lists the call.
with tracer.start_as_current_span("voice.call", fi_span_kind="conversation") as call:
    await session.start(agent=Assistant())
    ...

# agent.py     OpenTelemetry track
from observability.futureagi.setup import provider, tracer
from livekit.agents import AgentSession

session = AgentSession(stt=..., llm=..., tts=...)

with tracer.start_as_current_span("voice.call") as call:
    call.set_attribute("fi.span.kind", "CONVERSATION")
    await session.start(agent=Assistant())
    ...

  span opened before session.start()      conversation   ROOT    listed in the Voice tab
  span opened after  session.start()      conversation   child   listed nowhere

# agent.py, around the span from Step 3     Future AGI SDK track
with using_attributes(session_id=session_id, user_id=caller_id, tags=["prod"]):
    with tracer.start_as_current_span("voice.call", fi_span_kind="conversation") as call:
        await session.start(agent=Assistant())

# agent.py, around the span from Step 3     OpenTelemetry track
from observability.futureagi.setup import _scope

_scope.set({"session.id": session_id, "user.id": caller_id})
with tracer.start_as_current_span("voice.call") as call:
    call.set_attribute("fi.span.kind", "CONVERSATION")
    await session.start(agent=Assistant())

# observability/futureagi/voice_spans.py     both tracks
import json, uuid

DURATION = "call.duration"                    # seconds, number. Duration column, duration filter
TURNS = "call.total_turns"                    # number. Turns column, turn_count filter
TALK_RATIO = "call.talk_ratio"                # 0..1, number. Talk ratio filter
STATUS = "call.status"                        # provider status string
PHONE = "call.participant_phone_number"
PROVIDER = "gen_ai.system"                    # which parser the server uses for this call
TRANSCRIPT = "conversation.transcript"        # the whole thing, as JSON. What voice evals bind to
TRANSCRIPT_RENDERED = "fi.conversation.transcript"   # the same list. What the call drawer renders
RECORDING_MONO = "conversation.recording.mono.combined"
RECORDING_STEREO = "conversation.recording.stereo"

AGENT_ROLES = ("assistant", "agent", "bot")
CALLER_ROLES = ("user", "customer", "caller")


def _rows(turns):
    """turns is [(role, text), ...] in order, or [(role, text, start, duration), ...]
    where start is seconds from the beginning of the call and duration is how long
    that utterance took to speak."""
    return [tuple(t) + (None,) * (4 - len(t)) for t in turns]


def write_transcript(span, turns):
    """THREE keys, because three surfaces read it and none of them falls back to
    another. Write two of the three and the call looks complete on one screen and
    empty on the next.

      fi.conversation.transcript          the call drawer renders this one, and
                                          only this one, on a self-hosted agent
      conversation.transcript             what the eval variable picker resolves
      conversation.transcript.N.message.* what the error feed and the I/O panels walk

    The per-turn start and duration are what the Call Analytics strip computes
    Duration, Latency, User / AI and Silence from. Leave them out and those four
    cards read blank while Turns and Words are still right.
    """
    turns = _rows(turns)
    span.set_attribute(TRANSCRIPT_RENDERED, json.dumps(
        [{"id": str(uuid.uuid4()), "role": r, "content": c,
          "time": None if s is None else str(s), "duration": d}
         for r, c, s, d in turns]))
    span.set_attribute(TRANSCRIPT, json.dumps(
        [{"role": r, "content": c} for r, c, _, _ in turns]))
    for i, (role, text, _, _) in enumerate(turns):
        span.set_attribute("conversation.transcript.%d.message.role" % i, role)
        span.set_attribute("conversation.transcript.%d.message.content" % i, text)


def talk_ratio(turns):
    """Agent share of the words spoken. In an audio deployment use talk TIME."""
    turns = _rows(turns)
    agent = sum(len(c.split()) for r, c, _, _ in turns if r in AGENT_ROLES)
    total = sum(len(c.split()) for _, c, _, _ in turns) or 1
    return round(agent / total, 3)


def finish(span, *, turns, duration, provider, status="completed",
           phone=None, recording=None, stereo=None):
    """Close the conversation span with everything the Voice tab reads."""
    rows = _rows(turns)
    write_transcript(span, rows)
    span.set_attribute(TURNS, len(rows))
    span.set_attribute(DURATION, round(duration, 3))
    span.set_attribute(TALK_RATIO, talk_ratio(rows))
    span.set_attribute(PROVIDER, provider)
    span.set_attribute(STATUS, status)
    if phone:
        span.set_attribute(PHONE, phone)
    if recording:
        span.set_attribute(RECORDING_MONO, recording)
    if stereo:
        span.set_attribute(RECORDING_STEREO, stereo)
    span.set_attribute("input.value",
                       next((c for r, c, _, _ in rows if r in CALLER_ROLES), ""))
    span.set_attribute("output.value",
                       next((c for r, c, _, _ in reversed(rows) if r in AGENT_ROLES), ""))

export FI_VERIFY=1
export FI_VOICE_NO_RECORDING=1     # only if your deployment keeps no recording, per Step 5
export LLM_API_KEY="your-model-key"   # the agent's own provider key, not a Future AGI one

python observability/futureagi/fi_verify_voice.py preflight
python agent.py                    # pass your own asks as arguments to replace the two below
python observability/futureagi/fi_verify_voice.py check

The same six steps, run end to end

Everything above was run against LiveKit Agents 1.7.1 with traceai-livekit 0.1.1, on a project created for this page. This walkthrough is the Future AGI SDK track; the OpenTelemetry track writes the same attributes and is not repeated end to end. The whole run needs one model key, in LLM_API_KEY, and it is your model provider’s, never a Future AGI one. The listing calls Groq’s OpenAI-compatible endpoint by default because it serves both the STT and the LLM the example uses. Point it anywhere else with OPENAI_BASE_URL and AGENT_MODEL. No LiveKit account, no room, no phone number, no microphone, and no telephony spend, because AgentSession.run() is LiveKit’s own harness: it drives a real session with a real STT, a real LLM and a real turn, and session.start() takes no room.

pip install "livekit-agents[openai]" fi-instrumentation-otel traceai-livekit

# agent.py
"""One call, one conversation span, no room and no phone number.

AgentSession.run() is LiveKit's own harness: it drives a real session with no
room, no LiveKit account and no telephony, so this file is the whole worked
example and anyone can run it.
"""
import asyncio, os, sys, time, uuid

from observability.futureagi.setup import provider, tracer      # first import, before livekit
from observability.futureagi import voice_spans

from fi_instrumentation import using_attributes
from livekit.agents import Agent, AgentSession
from livekit.plugins import openai


class Assistant(Agent):
    def __init__(self):
        super().__init__(instructions=(
            "You are a voice assistant for an airline. Answer in one short spoken "
            "sentence, and never read out a list."))


async def main():
    # A call is a conversation, so the example is two exchanges, not one. Anything
    # you pass on the command line replaces them.
    asks = sys.argv[1:] or ["How much baggage can I bring?",
                            "And is a stroller counted separately?"]
    session_id = "call_" + uuid.uuid4().hex[:12]
    user_id = os.getenv("CALLER_ID", "acct_10427")

    base = os.getenv("OPENAI_BASE_URL", "https://api.groq.com/openai/v1")
    key = os.environ["LLM_API_KEY"]
    session = AgentSession(
        stt=openai.STT(model="whisper-large-v3-turbo", base_url=base, api_key=key),
        llm=openai.LLM(model=os.getenv("AGENT_MODEL", "openai/gpt-oss-120b"),
                       base_url=base, api_key=key),
    )

    started = time.monotonic()
    # The conversation span opens BEFORE session.start(). Opened after it, LiveKit's
    # own agent_session span is already the root, this one becomes a child, and the
    # Voice tab never lists the call: it selects a conversation span with no parent.
    with using_attributes(session_id=session_id, user_id=user_id, tags=["prod"]):
        with tracer.start_as_current_span("voice.call", fi_span_kind="conversation") as call:
            await session.start(agent=Assistant())
            for ask in asks:
                await session.run(user_input=ask, input_modality="text")
            turns = [(m.role, m.text_content) for m in session.history.items
                     if getattr(m, "role", None) in ("user", "assistant")
                     and getattr(m, "text_content", None)]
            voice_spans.finish(call, turns=turns,
                               duration=time.monotonic() - started,
                               provider="livekit")
            await session.aclose()

    provider.force_flush()
    for role, text in turns:
        print("  %-9s %s" % (role, text[:100]))
    print("\n  session.id = " + session_id)


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

input_modality="text" drives the turn through the LLM without synthesizing audio, which is what makes this runnable anywhere. Everything the gates check is identical on an audio session; only the transcript source changes, from session.history to whatever your STT emits.

What that produced

  PASS  V1   preflight ok, delivery ok
  PASS  V2   project_name='voice-cookbook-page-run' project_type='observe'
  PASS  V3   1 conversation span(s)
  PASS  V4   16 spans, 1 trace(s), 1 root(s), 0 orphan(s)
  PASS  V5   session.id='call_355937e9c818' user.id='acct_10427' on the conversation span
  PASS  V6   call.duration=1.119
  PASS  V7   call.total_turns=4 call.talk_ratio=0.727
  PASS  V8   4 turn(s) flattened; conversation.transcript present; fi.conversation.transcript present
  PASS  V9   provider='livekit'
  PASS  V10  no recording attribute, acknowledged: audio evals cannot bind to this call
  PASS  V11  4 LLM span(s), model on every one, prompt and completion on ['llm_node', 'llm_node']
  PASS  V12  no credential in any span attribute

  Future AGI sees this as a call
  GREEN LIGHT achieved

The call arrives with sixteen spans: the conversation span you wrote, and fifteen from LiveKit around it. llm_node and llm_request are the model call, agent_turn is one exchange, and the rest are session lifecycle.

One row per call. The row is the conversation span itself, which is why Step 3 has to open it before session.start(): the fifteen LiveKit spans are its children and never appear here on their own.

Open one and the product reads it as a call rather than a trace. The transcript, the turn count and the word count are the attributes Step 5 wrote, read straight back.

The call this page just produced, transcript and analytics, on call ID 1965f5ba.

Duration, Latency, User / AI and Silence read blank on purpose. Those four are computed from the per-turn time and duration on each transcript entry, which a text-mode run genuinely does not have. In an audio deployment, pass your STT’s utterance start and your TTS playback length as the third and fourth items of each turn, and they fill in.

Filter the attributes to transcript and all three keys are on the span, byte for byte as Step 5 wrote them.

fi.conversation.transcript renders the drawer, conversation.transcript is what a Traces-scope eval binds to, and the numbered keys drive the Messages panel and the error feed. Writing one does not fill the others in.

Read back through the product’s own endpoint, the same call comes out as a call rather than a trace:

  transcript_available = True   message_count = 4   turn_count = 4   talk_ratio = 0.727
    user       How much baggage can I bring?
    assistant  You may bring one checked bag up to 23 kg and one carry-on bag up ...
    user       And is a stroller counted separately?
    assistant  Yes ...

Every one of those fields is an attribute Step 5 wrote by name. Drop fi.conversation.transcript alone and the same call comes back with transcript_available: None and an empty transcript, with every other field unchanged.

Then move the conversation span two lines down, so it opens after session.start(), and run the identical agent again:

  FAIL  V3   1 conversation span(s), and ['voice.call'] has a parent, so the Voice tab will not list it

Eleven of twelve gates still pass. The trace is well formed, the transcript is complete, the duration is right, and the call cannot be found in the product. That is the whole reason this page has a checker.

The evals bound to it

Each of these binds only to an attribute the run above already carries.

Audio evals are the one family that will not bind to this run, because it produced no recording. That is exactly what V10 reports when you acknowledge it, and it is a real limit rather than a checker being lenient.

If your calls come from a managed provider

If your calls run on Vapi, Retell or Bland.ai, none of the six steps apply, because there is no code of yours in the path. You connect the provider once and Future AGI pulls each call and writes the conversation span for you, already typed, already parented at the root, already carrying the transcript, the recording URLs, the duration and the cost from the provider’s own payload.

Which means V2 through V12 are satisfied on arrival, and the only thing worth verifying is that calls are arriving at all. Run preflight to prove the project and the keys, then check the Voice tab after a real call completes. Some providers emit their call log at the end of the call rather than during it, so a call can arrive minutes after it happened.

The two tracks are not exclusive. A managed provider handles the telephony while your own tools and model calls run in your process, and instrumenting those with the six steps above gives you the child spans the provider’s payload cannot see.

If your agent is not Python

The listings translate line for line, and everything reaches the same endpoint over plain OpenTelemetry. Carry across:

• Both headers, X-Api-Key and X-Secret-Key, on https://api.futureagi.com/tracer/v1/traces
• project_name and project_type on the resource, not the span
• One conversation-typed span per call, with no parent, opened before the session starts
• The Step 5 attribute keys, byte for byte

The checker still runs. preflight and check are plain Python with no dependency at all, so they work next to an agent in any language. Only fi_verify_voice.attach() is Python-bound. Without it, write the capture yourself: one JSON object per line in .fi_verify/voice_spans.jsonl, each with name, trace_id, span_id, parent_id (null on the root), attrs, and resource.

Troubleshooting

Instrument the model calls inside the call with Instrument and Verify.