Agent Observatory

AI Agent Observability & Debugging Platform a self-hosted alternative to LangSmith / Helicone.

Instrument any Python AI agent in minutes. Every LLM call, tool invocation, and reasoning step gets captured, stored, and visualized in a live dashboard — including cost tracking, latency percentiles, hallucination risk scoring, reasoning drift detection, and a full execution DAG.

---

Screenshots

Dashboard	Execution DAG	Analytics
Live trace list, WebSocket feed, top-level metrics	Force-directed graph of every span, colored by kind and error state	Cost by model, latency percentiles, tool success rates

---

Features

Tracing

• Prompt traces — full input/output capture per LLM call (first 500 chars previewed)
• Tool calls — name, success/failure, latency per invocation
• Token usage — input + output tokens rolled up per trace and broken out per span
• Latency — wall-clock ms per span; p50/p95/p99 aggregated across all spans
• Retry chains — retry count tracked per span, visualized in the DAG

Quality Signals (experimental)

• Hallucination risk (0–1) — lightweight hedging-language density heuristic; replace with your own model-based scorer via span.hallucination_risk
• Reasoning drift (0–1) — Jaccard distance between consecutive LLM outputs; detects when reasoning direction shifts unexpectedly

Dashboard

• Execution DAG — D3 force-directed graph; drag, zoom, hover for details; collapses repeated retry nodes
• Span table — sortable, with kind badges, latency, cost, risk bars
• Cost analytics — total cost, breakdown by model and by agent, average per trace
• Latency stats — percentiles + average per span kind (llm / tool / chain / retrieval)
• Tool success % — per-tool call counts, success rate bar, average latency
• Prompt version history — register versioned prompts, diff content across versions
• Live events feed — WebSocket stream of span.created / trace.updated events in real time

---

Architecture

agent_observatory/
├── sdk/                    # Python SDK — instrument your agent
│   ├── __init__.py         # Public API exports
│   ├── tracer.py           # TraceContext, SpanContext, heuristics
│   ├── decorators.py       # @trace and @span decorators
│   └── client.py           # ObservatoryClient — async HTTP sender
├── server/                 # FastAPI backend
│   ├── main.py             # App entrypoint, CORS, lifespan
│   ├── db.py               # SQLAlchemy 2.0 async engine (SQLite)
│   ├── models.py           # ORM: Trace, Span, Metric, PromptVersion
│   ├── schemas.py          # Pydantic v2 ingest + response schemas
│   ├── broadcaster.py      # asyncio Queue WebSocket broadcaster
│   └── routes/
│       ├── ingest.py       # POST /ingest/traces|spans|prompts
│       ├── traces.py       # GET /traces, /traces/{id}, /traces/{id}/dag
│       ├── analytics.py    # GET /analytics/cost|latency|tools|prompts
│       └── ws.py           # WebSocket /ws/traces
├── dashboard/
│   └── index.html          # Single-file React 18 + D3 v7 dashboard
├── tests/                  # 30 pytest tests (unit + integration)
└── requirements.txt

Storage: SQLite (file-based, zero config). Swap to PostgreSQL by changing DATABASE_URL in server/db.py.

---

Quick Start

1. Install dependencies

cd agent_observatory
pip install -r requirements.txt

2. Start the server

uvicorn server.main:app --reload --port 7800

• Dashboard: http://localhost:7800
• API docs: http://localhost:7800/docs
• WebSocket: ws://localhost:7800/ws/traces

3. Instrument your agent

Using context managers

import asyncio
from sdk import ObservatoryClient, TraceContext, SpanContext

client = ObservatoryClient("http://localhost:7800")

async def run_agent():
    async with TraceContext("research-agent", agent_id="v2", client=client) as trace:

        # LLM call span
        async with SpanContext(
            "generate-plan",
            kind="llm",
            model="claude-sonnet-4-6",
            input_tokens=320,
            output_tokens=140,
            cost_usd=0.0028,
            latency_ms=950,
            trace_id=trace.trace_id,
            client=client,
        ) as span:
            response = await call_llm(prompt)
            span.output_text = response  # enables drift + hallucination scoring

        # Tool call span
        async with SpanContext(
            "web-search",
            kind="tool",
            tool_name="search",
            trace_id=trace.trace_id,
            client=client,
        ) as span:
            result = await search(query)
            span.tool_success = True
            span.output_text = str(result)

asyncio.run(run_agent())

Using decorators

from sdk import ObservatoryClient, trace, span

client = ObservatoryClient("http://localhost:7800")

@trace(name="research-agent", agent_id="v2", client=client)
async def run_agent(query: str) -> str:
    plan = await generate_plan(query)
    result = await search_web(plan)
    return result

@span(name="generate-plan", kind="llm", model="claude-sonnet-4-6", client=client)
async def generate_plan(query: str) -> str:
    ...

@span(name="web-search", kind="tool", tool_name="search", client=client)
async def search_web(plan: str) -> str:
    ...

Register prompt versions

await client.register_prompt(
    name="system-prompt",
    version="2.1",
    content="You are a research assistant...",
    author="rakesh",
    description="Added citation instructions",
)

---

API Reference

Ingest

Method	Path	Description
POST	/ingest/traces	Open a new trace
PATCH	/ingest/traces/{id}	Update / close a trace
POST	/ingest/spans	Record a span
POST	/ingest/prompts	Register a prompt version

Query

Method	Path	Description
GET	/traces	Paginated trace list (?status=&agent_id=&limit=&offset=)
GET	/traces/{id}	Full trace + all spans
GET	/traces/{id}/dag	Execution DAG nodes + edges
DELETE	/traces/{id}	Hard delete trace + spans

Analytics

Method	Path	Description
GET	/analytics/cost	Total cost, by model, by agent
GET	/analytics/latency	p50/p95/p99 + avg per kind
GET	/analytics/tools	Tool success rates + latency
GET	/analytics/prompts	Prompt version history (?name=)

WebSocket

ws://localhost:7800/ws/traces

Events:

{"event": "span.created", "data": {"id": "...", "name": "...", "latency_ms": 800, "cost_usd": 0.002}}
{"event": "trace.updated", "data": {"id": "...", "status": "completed", "total_cost_usd": 0.012}}
{"event": "ping"}

---

Span Schema

Field	Type	Description
id	str	Unique span ID (UUID)
trace_id	str	Parent trace ID
parent_span_id	str|null	Parent span for nesting
name	str	Human-readable step name
kind	str	llm | tool | chain | retrieval | other
model	str|null	Model identifier
input_tokens	int	Prompt tokens
output_tokens	int	Completion tokens
cost_usd	float	USD cost for this span
latency_ms	int|null	Wall-clock duration
tool_name	str|null	Tool name (kind=tool)
tool_success	bool|null	Tool outcome
hallucination_risk	float 0–1	Hedging-language density score
reasoning_drift	float 0–1	Jaccard distance from prior output
prompt_version	str|null	Prompt version tag
retry_count	int	Number of retries
error	str|null	Exception message if failed
input_preview	str|null	First 500 chars of input
output_preview	str|null	First 500 chars of output

---

Configuration

Environment Variable	Default	Description
OBSERVATORY_DB	observatory.db	SQLite file path

To use PostgreSQL:

# server/db.py
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/observatory"

Add asyncpg to requirements.txt.

---

Running Tests

pytest tests/ -v
# 30 tests: ingest, analytics, SDK unit tests

pytest tests/ --cov=server --cov=sdk --cov-report=term-missing

---

Comparison

Feature	Agent Observatory	LangSmith	Helicone
Self-hosted	✅	❌ (cloud)	❌ (cloud)
Execution DAG	✅	✅	❌
Reasoning drift	✅	❌	❌
Hallucination risk	✅	❌	❌
Prompt versioning	✅	✅	✅
WebSocket live stream	✅	✅	❌
Zero infra (SQLite)	✅	❌	❌
Open source	✅	❌	partial

---

Tech Stack

• Backend: FastAPI + SQLAlchemy 2.0 (async) + aiosqlite
• Schemas: Pydantic v2
• Real-time: WebSockets (asyncio Queue per client)
• Dashboard: React 18 + D3 v7 (single HTML file, no build step)
• Tests: pytest + pytest-asyncio + httpx AsyncClient

---

License

MIT