A lightweight observability layer for AI/LLM applications with tracing, metrics, and dashboards.
Production AI systems are black boxes. When something goes wrong, you need answers:
- Why was this response slow?
- Which agent in my workflow failed?
- How much is this costing me?
- Is my RAG system retrieving the right documents?
AI Observe provides the tools to debug, monitor, and improve AI systems in production.
# Install
pip install -e .
# Add tracing to your code
from ai_observe import trace
@trace(name="answer_question")
def answer_question(question: str) -> str:
return llm.complete(question)
# Start the dashboard
python -m ai_observe.dashboard
# Open http://localhost:8080from ai_observe import trace, trace_workflow
@trace(name="my_llm_call", model="llama3.2")
def call_llm(prompt: str) -> str:
response = ollama.chat(model="llama3.2", messages=[...])
return response["message"]["content"]
# Automatically captures:
# - Latency
# - Input/output previews
# - Token usage (if available)
# - Errorsfrom ai_observe import trace_workflow
@trace_workflow(name="research_agent_workflow")
def run_research_crew():
# Your CrewAI/LangChain workflow
# Each agent step is captured as a span
# Shows: Agent A → Agent B → Agent C flow
return crew.kickoff()# Ollama
from ai_observe.integrations import OllamaIntegration
OllamaIntegration.patch()
# All Ollama calls are now automatically traced
# CrewAI
from ai_observe.integrations import CrewAIIntegration
CrewAIIntegration.patch()
# Crew workflows are now traced
# LangChain
from ai_observe.integrations import LangChainIntegration
handler = LangChainIntegration.get_callback_handler()
llm.invoke("Hello", config={"callbacks": [handler]})┌─────────────────────────────────────────────────────────────────┐
│ AI Observe Dashboard Last 24 hrs │
├─────────────────────────────────────────────────────────────────┤
│ Requests: 1,247 Avg Latency: 2.3s Errors: 0.4% │
│ Tokens: 892K Est Cost: $4.28 P99: 8.1s │
├─────────────────────────────────────────────────────────────────┤
│ Recent Traces │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ [12:34:56] research_workflow ████████░░ 4.2s $0.02 │ │
│ │ └─ researcher_agent ████░░░░░░ 1.8s │ │
│ │ └─ writer_agent ███░░░░░░░ 1.2s │ │
│ │ └─ editor_agent ██░░░░░░░░ 0.9s │ │
│ └──────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────┐
│ Your Application │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ @trace │ │ CrewAI │ │LangChain │ │ Ollama │ │
│ │decorator │ │ patch │ │ callback │ │ patch │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │ │
│ └──────────────┴──────────────┴──────────────┘ │
│ │ │
│ ┌─────▼─────┐ │
│ │ Tracer │ │
│ │ (spans, │ │
│ │ traces) │ │
│ └─────┬─────┘ │
├──────────────────────────────┼──────────────────────────────────┤
│ │ │
│ ┌──────────────────────────┼──────────────────────────────┐ │
│ │ ┌─────▼─────┐ │ │
│ │ │ Collectors│ │ │
│ │ ┌───────────────┼───────────┼───────────────┐ │ │
│ │ │ │ │ │ │ │
│ │ ┌──▼───┐ ┌───▼───┐ ┌───▼───┐ ┌───▼───┐ │ │
│ │ │Latency│ │Tokens │ │ Cost │ │Errors │ │ │
│ │ └───────┘ └───────┘ └───────┘ └───────┘ │ │
│ │ │ │
│ └────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────┼──────────────────────────────┐ │
│ │ ┌─────▼─────┐ │ │
│ │ │ Storage │ │ │
│ │ ┌────────┴───────────┴────────┐ │ │
│ │ │ │ │ │
│ │ ┌───▼───┐ ┌────▼────┐ │ │
│ │ │SQLite │ │PostgreSQL│ │ │
│ │ └───────┘ └──────────┘ │ │
│ │ │ │
│ └────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────┼──────────────────────────────┐ │
│ │ ┌─────▼─────┐ │ │
│ │ │ Dashboard │ │ │
│ │ │ (FastAPI) │ │ │
│ │ └───────────┘ │ │
│ └────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
from ai_observe import trace, Tracer
# Method 1: Decorator
@trace(name="chat_response")
def get_chat_response(message: str) -> str:
return llm.chat(message)
# Method 2: Context manager
tracer = Tracer.get_instance()
with tracer.span("my_operation") as span:
result = do_something()
span.output_preview = result[:500]
# Method 3: Manual spans
tracer = Tracer.get_instance()
span = tracer.start_span("manual_span")
try:
result = do_something()
tracer.end_span("success")
except Exception as e:
tracer.end_span("error", str(e))from ai_observe.integrations import CrewAIIntegration
from ai_observe import trace_workflow
from crewai import Crew, Agent, Task
# Enable automatic tracing
CrewAIIntegration.patch()
# Or use decorator for explicit tracing
@trace_workflow(name="research_crew")
def run_research():
researcher = Agent(
role="Researcher",
goal="Find information",
backstory="Expert researcher"
)
writer = Agent(
role="Writer",
goal="Write content",
backstory="Expert writer"
)
crew = Crew(
agents=[researcher, writer],
tasks=[Task(description="Research AI trends", agent=researcher)]
)
return crew.kickoff()Start the dashboard:
from ai_observe.dashboard import run_dashboard
run_dashboard(host="0.0.0.0", port=8080)Views:
- Overview: Requests/min, avg latency, error rate, total cost
- Traces: Drill into individual requests, see full conversation
- Timeline: Visual span timeline for each trace
- Details: Input/output previews, error messages
from ai_observe.storage import SQLiteStorage
storage = SQLiteStorage("traces.db")
await storage.initialize()from ai_observe.storage import PostgresStorage
storage = PostgresStorage(
host="localhost",
database="ai_observe",
user="postgres",
password="secret"
)
await storage.initialize()Track specific metrics:
from ai_observe.collectors import (
LatencyCollector,
TokenCollector,
CostCollector,
ErrorCollector,
)
# Latency tracking
latency = LatencyCollector()
latency.record(150.0, model="llama3.2", operation="chat")
stats = latency.get_stats()
print(f"P95 latency: {stats.p95_ms}ms")
# Cost tracking
cost = CostCollector()
cost.record(
prompt_tokens=100,
completion_tokens=50,
model="gpt-4o-mini"
)
print(f"Estimated monthly: ${cost.estimate_monthly_cost():.2f}")| Feature | AI Observe | LangSmith | Weights & Biases |
|---|---|---|---|
| Self-hosted | Yes | No | No |
| Local-first | Yes | No | No |
| Simple decorator API | Yes | Limited | No |
| Multi-agent tracing | Yes | Yes | Limited |
| Cost tracking | Yes | Limited | No |
| Open source | Yes | No | Partial |
- OpenTelemetry - Span-based tracing concepts
- Anthropic's internal tooling - Focus on debugging AI systems
- LangSmith - LLM-specific observability patterns
Contributions welcome! Areas of interest:
- Additional framework integrations
- Dashboard improvements
- Advanced analytics
- Performance optimizations
MIT