LLM Telemetry Toolkit

Use Case Synopsis

llm-telemetry-toolkit is a production-focused telemetry layer for LLM agents and tools.
It captures structured interactions with async queue writes (or true sync writes when needed), sanitizes content, and emits JSON/Markdown/CSV logs for local debugging, analytics, and release-grade observability.

Installation

pip install llm-telemetry-toolkit

Optional OpenTelemetry export support:

pip install "llm-telemetry-toolkit[otel]"

Integration Guide

1) Standalone usage

from llm_telemetry_toolkit import LLMInteraction, LLMLogger, TelemetryConfig

config = TelemetryConfig(
    session_id="demo_session",
    output_formats=["json", "md", "csv"],
    ensure_ascii=False,
)
logger = LLMLogger(config)

result = logger.log(
    LLMInteraction(
        session_id="demo_session",
        model_name="my-model",
        prompt="Explain Rayleigh scattering in one sentence.",
        response="The sky looks blue because shorter blue wavelengths scatter more in the atmosphere.",
        response_time_seconds=0.52,
    ),
    sync=True,
)

print(result.model_dump())
logger.shutdown()

2) Agentic / MCP usage

Use this package as your telemetry boundary around tool execution:

from llm_telemetry_toolkit import LLMInteraction, LLMLogger, SessionContext, TelemetryConfig

logger = LLMLogger(TelemetryConfig(session_id="agent_run_001", mask_pii=True))

def run_tool(tool_name: str, tool_input: str, tool_output: str, latency: float) -> None:
    logger.log(
        LLMInteraction(
            session_id="agent_run_001",
            model_name="deepseek:r1",
            interaction_type="tool_call",
            tool_name=tool_name,
            prompt=tool_input,
            response=tool_output,
            response_time_seconds=latency,
            metadata={"trace_layer": "mcp_server"},
        )
    )

with SessionContext("user_session_abc"):
    run_tool("web_scraper", "https://example.com", "200 OK", 0.23)

logger.shutdown()

3) Ollama adapters (sync + async, package-native)

import asyncio
from llm_telemetry_toolkit import AsyncOllamaClient, OllamaClient, OllamaTransportConfig

transport = OllamaTransportConfig(max_retries=2, backoff_initial_seconds=0.2)

sync_client = OllamaClient("http://localhost:11434", transport_config=transport)
print(sync_client.list_models())
sync_client.close()

async def run() -> None:
    async with AsyncOllamaClient("http://localhost:11434", transport_config=transport) as client:
        payload = await client.chat(
            model="tinyllama:latest",
            messages=[{"role": "user", "content": "hello"}],
            stream=False,
        )
        print(payload)

asyncio.run(run())

4) Ollama streaming chunk API (sync + async)

import asyncio
from llm_telemetry_toolkit import AsyncOllamaClient, OllamaClient

with OllamaClient("http://localhost:11434") as client:
    for event in client.stream_generate("tinyllama:latest", "Count 1 to 3"):
        print(event)

async def run_stream() -> None:
    async with AsyncOllamaClient("http://localhost:11434") as client:
        async for event in client.stream_chat(
            model="tinyllama:latest",
            messages=[{"role": "user", "content": "Say hello in 3 words"}],
        ):
            print(event)

asyncio.run(run_stream())

5) Optional OpenTelemetry export

from llm_telemetry_toolkit import LLMLogger, TelemetryConfig

logger = LLMLogger(
    TelemetryConfig(
        session_id="otel_demo",
        enable_otel_export=True,
        otel_tracer_name="research_tool",
        otel_span_name="llm.interaction",
        otel_service_name="llm-telemetry-toolkit",
        otel_auto_configure=True,
        otel_otlp_endpoint="http://localhost:4318/v1/traces",
        otel_sampler_ratio=1.0,
    )
)

Note: OTel export is additive to file logging and uses opentelemetry-api/sdk + OTLP exporter when auto-configure is enabled.

Commands & Arguments

CLI entry points

• llm-telemetry-toolkit ...
• python -m llm_telemetry_toolkit.interface.cli ...

Commands

view

Render recent interactions for a session.

llm-telemetry-toolkit view --session <SESSION_ID> [--dir ./logs] [--limit 5]

Arguments:

• --session (required): session id folder under llm_interactions/
• --dir (optional): base log directory (default: ./logs)
• --limit (optional): number of latest interactions to display (default: 5)

stats

Aggregate totals (latency, token counts, cost, model usage).

llm-telemetry-toolkit stats --session <SESSION_ID> [--dir ./logs]

Arguments:

• --session (required): session id folder under llm_interactions/
• --dir (optional): base log directory (default: ./logs)

Configuration (TelemetryConfig)

Structured interaction fields (LLMInteraction)

In addition to free-form metadata, the schema now supports explicit optional fields for downstream consistency:

• Chat/tooling: request_messages, response_message_role, finish_reason, tool_calls
• Embeddings: embedding_input_count, embedding_vector_count, embedding_dimensions
• Typed helper models: ChatMessage, ToolCall, ToolFunctionCall

Examples

Decorator-based logging (sync + async)

import asyncio
from llm_telemetry_toolkit import LLMLogger, TelemetryConfig, monitor_interaction

logger = LLMLogger(TelemetryConfig(session_id="decorator_demo"))

@monitor_interaction(logger, interaction_type="sync_call")
def multiply(x: int, y: int) -> int:
    return x * y

@monitor_interaction(logger, interaction_type="async_call")
async def async_multiply(x: int, y: int) -> int:
    await asyncio.sleep(0.05)
    return x * y

print(multiply(3, 4))
print(asyncio.run(async_multiply(5, 6)))
logger.shutdown()

True sync writes (for strict confirmation paths)

from llm_telemetry_toolkit import LLMInteraction, LLMLogger, TelemetryConfig

logger = LLMLogger(TelemetryConfig(session_id="sync_demo"))
result = logger.log(
    LLMInteraction(
        session_id="sync_demo",
        model_name="sync-model",
        prompt="hello",
        response="world",
        response_time_seconds=0.1,
    ),
    sync=True,
)
print(result.success, result.queued, result.write_confirmed, result.created_files)
logger.shutdown()

Testing

Standard local test run

python -B -m pytest -q

Optional live Ollama integration test (private/local only)

1. Copy settings.example.cfg to settings.cfg.
2. Update settings.cfg with your private local server values.
3. Run:

$env:OLLAMA_INTEGRATION = "1"
python -m pytest tests/test_ollama_integration.py -q

Packaging & Release Notes

• Versioning is VCS-driven via hatch-vcs and git tags (vX.Y.Z).
• Console script entry point is published as llm-telemetry-toolkit.
• PyPI publish workflow is tag-triggered (push on v* tags).
• Optional TestPyPI publish can be run manually via GitHub Actions workflow dispatch.

Recommended release sequence:

1. Dry run preflight:

python .\tools\publish_release.py --project .\llm-telemetry-toolkit --version 0.2.0 --dry-run

2. Real release tag push:

python .\tools\publish_release.py --project .\llm-telemetry-toolkit --version 0.2.0

3. (Optional) Manual TestPyPI publish from Actions UI:
   • Workflow: Publish to PyPI
   • Input: publish_target = testpypi

Author & Links

Created by Roy Dawson IV

• GitHub: https://github.com/imyourboyroy
• PyPi: https://pypi.org/user/ImYourBoyRoy/

License

MIT License