agentreplay 0.1.4

Agentreplay is a high-performance observability and tracing platform for LLM agents and AI apps, combining semantic search, specialized evals (RAGAS, G-Eval, toxicity), and Git-like versioning of prompts and responses.

Agentreplay Python SDK

The observability platform for LLM agents and AI applications. Trace every LLM call, tool invocation, and agent step with minimal code changes.

---

✨ Features

Feature	Description
🚀 Zero-Config Setup	Works out of the box with environment variables
🎯 One-Liner Instrumentation	Wrap OpenAI/Anthropic clients in one line
🔧 Decorator-Based Tracing	@traceable for any function
🔄 Async Native	Full support for async/await patterns
🔒 Privacy First	Built-in PII redaction and scrubbing
📊 Token Tracking	Automatic token usage capture
🌐 Framework Agnostic	Works with LangChain, LlamaIndex, CrewAI, etc.
⚡ Batched Transport	Efficient background sending with retry

---

📦 Installation

# Basic installation (minimal dependencies)
pip install agentreplay

# With OpenTelemetry support
pip install agentreplay[otel]

# With LangChain integration
pip install agentreplay[langchain]

# With LangGraph integration
pip install agentreplay[langgraph]

# With LlamaIndex integration
pip install agentreplay[llamaindex]

# Full installation (all integrations)
pip install agentreplay[otel,langchain,langgraph,llamaindex]

---

🚀 Quick Start

1. Set Environment Variables

export AGENTREPLAY_API_KEY="your-api-key"
export AGENTREPLAY_PROJECT_ID="my-project"
# Optional
export AGENTREPLAY_BASE_URL="https://api.agentreplay.io"

2. Enable Auto-Instrumentation

After installing the package, run the install command to enable zero-code auto-instrumentation:

# Run this once after pip install
agentreplay-install

This installs a .pth file that automatically initializes Agentreplay when Python starts, enabling automatic tracing of OpenAI, Anthropic, and other LLM libraries without any code changes.

Option A: Zero-Code (after running agentreplay-install)

# Just run your script - instrumentation happens automatically!
python my_app.py

Option B: Code-Based Initialization

Add these two lines at the very beginning of your main file, before any other imports:

import agentreplay
agentreplay.init()  # Must be called before importing OpenAI, Anthropic, etc.

# Now import your LLM clients - they're automatically traced!
from openai import OpenAI

client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello!"}]
)
# All OpenAI calls are automatically traced!

# Ensure traces are sent before exit
agentreplay.flush()

⚠️ Important: agentreplay.init() must be called before importing OpenAI, Anthropic, or other LLM libraries for auto-instrumentation to work.

Managing Auto-Instrumentation

# Install auto-instrumentation (run once after pip install)
agentreplay-install

# Check if installed
agentreplay-install --check

# Uninstall auto-instrumentation
agentreplay-install --uninstall

3. Manual Tracing with Decorators

import agentreplay

# Initialize (reads from env vars automatically)
agentreplay.init()

# Trace any function with a simple decorator
@agentreplay.traceable
def my_ai_function(query: str) -> str:
    # Your AI logic here
    return f"Response to: {query}"

# Call your function - it's automatically traced!
result = my_ai_function("What is the capital of France?")

# Ensure all traces are sent before exit
agentreplay.flush()

That's it! Your function calls are now being traced and sent to Agentreplay.

---

🔧 Core API Reference

Initialization

import agentreplay

# Option 1: Environment variables (recommended for production)
agentreplay.init()

# Option 2: Explicit configuration
agentreplay.init(
    api_key="your-api-key",
    project_id="my-project",
    base_url="https://api.agentreplay.io",
    
    # Optional settings
    tenant_id="default",       # Multi-tenant identifier
    agent_id="default",        # Default agent ID
    enabled=True,              # Set False to disable in tests
    capture_input=True,        # Capture function inputs
    capture_output=True,       # Capture function outputs
    batch_size=100,            # Batch size before sending
    flush_interval=5.0,        # Auto-flush interval in seconds
    debug=False,               # Enable debug logging
)

# Check if SDK is initialized
from agentreplay.sdk import is_initialized
if is_initialized():
    print("SDK is ready!")

# Get current configuration
config = agentreplay.sdk.get_config()
print(f"Project: {config.project_id}")

---

🎯 The @traceable Decorator

The primary and most Pythonic way to instrument your code:

Basic Usage

from agentreplay import traceable

# Just add the decorator - that's it!
@traceable
def process_query(query: str) -> str:
    return call_llm(query)

# Works with any function signature
@traceable
def complex_function(data: dict, *, optional_param: str = "default") -> list:
    return process(data, optional_param)

With Options

from agentreplay import traceable, SpanKind

# Custom span name and kind
@traceable(name="openai_chat", kind=SpanKind.LLM)
def call_openai(messages: list) -> str:
    return openai_client.chat.completions.create(
        model="gpt-4",
        messages=messages
    )

# Disable input capture for sensitive data
@traceable(capture_input=False)
def authenticate(password: str) -> bool:
    return verify_password(password)

# Disable output capture
@traceable(capture_output=False)
def get_secret() -> str:
    return fetch_secret_from_vault()

# Add static metadata
@traceable(metadata={"version": "2.0", "model": "gpt-4", "team": "ml"})
def enhanced_query(query: str) -> str:
    return process(query)

Async Functions

Full async/await support - no changes needed:

import asyncio
from agentreplay import traceable, SpanKind

@traceable(kind=SpanKind.LLM)
async def async_llm_call(prompt: str) -> str:
    response = await openai_client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

@traceable
async def process_batch(queries: list[str]) -> list[str]:
    # All concurrent calls are traced with proper parent-child relationships
    tasks = [async_llm_call(q) for q in queries]
    return await asyncio.gather(*tasks)

# Run
results = asyncio.run(process_batch(["query1", "query2", "query3"]))

---

📐 Context Manager: trace()

For more control over span attributes and timing:

from agentreplay import trace, SpanKind

def complex_operation(query: str) -> dict:
    with trace("process_query", kind=SpanKind.CHAIN) as span:
        # Set input data
        span.set_input({"query": query, "timestamp": time.time()})
        
        # Nested span for document retrieval
        with trace("retrieve_documents", kind=SpanKind.RETRIEVER) as retriever_span:
            docs = vector_db.search(query, top_k=5)
            retriever_span.set_output({"document_count": len(docs)})
            retriever_span.set_attribute("vector_db", "pinecone")
        
        # Nested span for LLM generation
        with trace("generate_response", kind=SpanKind.LLM) as llm_span:
            llm_span.set_model("gpt-4", provider="openai")
            response = generate_response(query, docs)
            llm_span.set_token_usage(
                prompt_tokens=150,
                completion_tokens=200,
                total_tokens=350
            )
            llm_span.set_output({"response_length": len(response)})
        
        # Add events for debugging
        span.add_event("processing_complete", {"doc_count": len(docs)})
        
        # Set final output
        span.set_output({"response": response, "sources": len(docs)})
        
        return {"response": response, "sources": docs}

Manual Span Control

For cases where you need explicit control:

from agentreplay import start_span, SpanKind

def long_running_operation():
    span = start_span("background_job", kind=SpanKind.TOOL)
    span.set_input({"job_type": "data_sync"})
    
    try:
        # Long running work...
        for i in range(100):
            process_item(i)
            if i % 10 == 0:
                span.add_event("progress", {"completed": i})
        
        span.set_output({"items_processed": 100})
        
    except Exception as e:
        span.set_error(e)
        raise
        
    finally:
        span.end()  # Always call end()

---

🔌 LLM Client Wrappers

OpenAI (Recommended)

One line to instrument all OpenAI calls:

from openai import OpenAI
from agentreplay import init, wrap_openai, flush

init()

# Wrap the client - all calls are now traced automatically!
client = wrap_openai(OpenAI())

# Use normally - tracing happens in the background
response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain quantum computing in simple terms."}
    ],
    temperature=0.7,
)

print(response.choices[0].message.content)

# Embeddings are traced too
embedding = client.embeddings.create(
    model="text-embedding-ada-002",
    input="Hello world"
)

flush()

Async OpenAI

from openai import AsyncOpenAI
from agentreplay import init, wrap_openai

init()

# Works with async client too
async_client = wrap_openai(AsyncOpenAI())

async def main():
    response = await async_client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "Hello!"}]
    )
    return response.choices[0].message.content

Anthropic

from anthropic import Anthropic
from agentreplay import init, wrap_anthropic, flush

init()

# Wrap the Anthropic client
client = wrap_anthropic(Anthropic())

# Use normally
message = client.messages.create(
    model="claude-3-opus-20240229",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Explain the theory of relativity."}
    ]
)

print(message.content[0].text)
flush()

Disable Content Capture

For privacy-sensitive applications:

# Don't capture message content, only metadata
client = wrap_openai(OpenAI(), capture_content=False)

# Traces will still include:
# - Model name
# - Token counts
# - Latency
# - Error information
# But NOT the actual messages or responses

---

🏷️ Context Management

Global Context

Set context that applies to ALL subsequent traces:

from agentreplay import set_context, get_global_context, clear_context

# Set user context (persists until cleared)
set_context(
    user_id="user-123",
    session_id="session-456",
    agent_id="support-bot",
)

# Add more context later
set_context(
    environment="production",
    version="1.2.0",
    region="us-west-2",
)

# Get current global context
context = get_global_context()
print(context)  # {'user_id': 'user-123', 'session_id': 'session-456', ...}

# Clear all context
clear_context()

Request-Scoped Context

For web applications with per-request context:

from agentreplay import with_context

async def handle_api_request(request):
    # Context only applies within this block
    with with_context(
        user_id=request.user_id,
        request_id=request.headers.get("X-Request-ID"),
        ip_address=request.client_ip,
    ):
        # All traces here include this context
        result = await process_request(request)
        
    # Context automatically cleared after block
    return result

# Async version works too
async def async_handler(request):
    async with with_context(user_id=request.user_id):
        return await async_process(request)

Multi-Agent Context

For multi-agent systems like CrewAI or AutoGen:

from agentreplay import AgentContext

def run_multi_agent_workflow():
    # Each agent gets its own context
    with AgentContext(
        agent_id="researcher",
        session_id="workflow-123",
        workflow_id="content-creation",
    ):
        # All LLM calls here are tagged with agent_id="researcher"
        research_results = researcher_agent.run()
    
    with AgentContext(
        agent_id="writer",
        session_id="workflow-123",  # Same session
        workflow_id="content-creation",
    ):
        # These calls are tagged with agent_id="writer"
        article = writer_agent.run(research_results)
    
    with AgentContext(
        agent_id="editor",
        session_id="workflow-123",
        workflow_id="content-creation",
    ):
        final_article = editor_agent.run(article)
    
    return final_article

---

🔒 Privacy & Data Redaction

Configure Privacy Settings

from agentreplay import configure_privacy

configure_privacy(
    # Enable built-in patterns for common PII
    use_builtin_patterns=True,  # Emails, credit cards, SSNs, phones, API keys
    
    # Add custom regex patterns
    redact_patterns=[
        r"secret-\w+",           # Custom secret format
        r"internal-id-\d+",      # Internal IDs
        r"password:\s*\S+",      # Password fields
    ],
    
    # Completely scrub these JSON paths
    scrub_paths=[
        "input.password",
        "input.credentials.api_key",
        "output.user.ssn",
        "metadata.internal_token",
    ],
    
    # Hash PII instead of replacing with [REDACTED]
    # Allows tracking unique values without exposing data
    hash_pii=True,
    hash_salt="your-secret-salt-here",
    
    # Custom replacement text
    redacted_text="[SENSITIVE]",
)

Built-in Redaction Patterns

The SDK includes patterns for:

Type	Example	Redacted As
Email	[e-mail]	[REDACTED]
Credit Card	4111-1111-1111-1111	[REDACTED]
SSN	123-45-6789	[REDACTED]
Phone (US)	[phone]	[REDACTED]
Phone (Intl)	[phone]	[REDACTED]
API Key	sk-proj-abc123...	[REDACTED]
Bearer Token	Bearer eyJ...	[REDACTED]
JWT	eyJhbG...	[REDACTED]
IP Address	192.168.1.1	[REDACTED]

Manual Redaction

from agentreplay import redact_payload, redact_string, hash_pii

# Redact an entire payload
data = {
    "user": {
        "email": "[e-mail]",
        "phone": "[phone]",
    },
    "message": "My credit card is 4111-1111-1111-1111",
    "api_key": "sk-proj-abcdefghijk",
}

safe_data = redact_payload(data)
# Result:
# {
#     "user": {
#         "email": "[REDACTED]",
#         "phone": "[REDACTED]",
#     },
#     "message": "My credit card is [REDACTED]",
#     "api_key": "[REDACTED]",
# }

# Redact a single string
safe_message = redact_string("Contact me at [e-mail]")
# "Contact me at [REDACTED]"

# Hash for consistent anonymization (same input = same hash)
user_hash = hash_pii("[e-mail]")
# "[HASH:a1b2c3d4]"

# Useful for analytics without exposing PII
print(f"User {user_hash} performed action")

Temporary Privacy Context

from agentreplay.privacy import privacy_context

# Add extra redaction rules for a specific block
with privacy_context(
    redact_patterns=[r"internal-token-\w+"],
    scrub_paths=["metadata.debug_info"],
):
    # These rules only apply within this block
    result = process_sensitive_data(data)

# Original rules restored after block

---

📊 Span Kinds

Use semantic span kinds for better visualization and filtering:

from agentreplay import SpanKind

# Available span kinds
SpanKind.CHAIN       # Orchestration, workflows, pipelines
SpanKind.LLM         # LLM API calls (OpenAI, Anthropic, etc.)
SpanKind.TOOL        # Tool/function calls, actions
SpanKind.RETRIEVER   # Vector DB search, document retrieval
SpanKind.EMBEDDING   # Embedding generation
SpanKind.GUARDRAIL   # Safety checks, content filtering
SpanKind.CACHE       # Cache operations
SpanKind.HTTP        # HTTP requests
SpanKind.DB          # Database queries

Example usage:

from agentreplay import traceable, SpanKind

@traceable(kind=SpanKind.RETRIEVER)
def search_documents(query: str) -> list:
    return vector_db.similarity_search(query, k=5)

@traceable(kind=SpanKind.LLM)
def generate_answer(query: str, docs: list) -> str:
    return llm.generate(query, context=docs)

@traceable(kind=SpanKind.CHAIN)
def rag_pipeline(query: str) -> str:
    docs = search_documents(query)
    return generate_answer(query, docs)

---

⚙️ Lifecycle Management

Flushing Traces

Always ensure traces are sent before your application exits:

import agentreplay

agentreplay.init()

# Your application code...

# Option 1: Manual flush with timeout
agentreplay.flush(timeout=10.0)  # Wait up to 10 seconds

# Option 2: Full graceful shutdown
agentreplay.shutdown(timeout=30.0)  # Flush and cleanup

# Option 3: Auto-registered (init() registers atexit handler automatically)
# Traces are flushed on normal program exit

Serverless / AWS Lambda

Critical: Always flush explicitly before the function returns!

import agentreplay

agentreplay.init()

@agentreplay.traceable
def process_event(event):
    # Your logic here
    return {"processed": True}

def lambda_handler(event, context):
    try:
        result = process_event(event)
        return {
            "statusCode": 200,
            "body": json.dumps(result)
        }
    finally:
        # CRITICAL: Flush before Lambda freezes
        agentreplay.flush(timeout=5.0)

FastAPI / Starlette

from fastapi import FastAPI
from contextlib import asynccontextmanager
import agentreplay

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    agentreplay.init()
    yield
    # Shutdown
    agentreplay.shutdown(timeout=10.0)

app = FastAPI(lifespan=lifespan)

@app.post("/chat")
async def chat(request: ChatRequest):
    # Traces are sent automatically in background
    return await process_chat(request)

Diagnostics

import agentreplay

# Get SDK statistics
stats = agentreplay.get_stats()
print(f"Spans sent: {stats.get('spans_sent', 0)}")
print(f"Spans pending: {stats.get('spans_pending', 0)}")
print(f"Errors: {stats.get('errors', 0)}")
print(f"Batches sent: {stats.get('batches_sent', 0)}")

# Health check - verify backend connectivity
if agentreplay.ping():
    print("✅ Backend is reachable")
else:
    print("❌ Cannot reach backend")

---

🔗 Framework Integrations

LangChain

from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
import agentreplay

agentreplay.init()

@agentreplay.traceable(name="langchain_qa", kind=agentreplay.SpanKind.CHAIN)
def answer_question(question: str) -> str:
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    response = llm.invoke([HumanMessage(content=question)])
    return response.content

result = answer_question("What is machine learning?")
agentreplay.flush()

LangGraph

from langgraph.graph import StateGraph, END
import agentreplay

agentreplay.init()

@agentreplay.traceable(name="agent_node", kind=agentreplay.SpanKind.LLM)
def agent_node(state):
    # Agent logic
    response = llm.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

@agentreplay.traceable(name="tool_node", kind=agentreplay.SpanKind.TOOL)
def tool_node(state):
    # Tool execution
    result = execute_tool(state["tool_call"])
    return {"messages": state["messages"] + [result]}

# Build graph with traced nodes
workflow = StateGraph(State)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
# ... rest of graph definition

CrewAI

from crewai import Agent, Task, Crew
import agentreplay

agentreplay.init()

# Wrap the LLM client
wrapped_llm = agentreplay.wrap_openai(OpenAI())

# Track each agent with context
with agentreplay.AgentContext(agent_id="researcher", workflow_id="article-creation"):
    researcher = Agent(
        role="Senior Researcher",
        goal="Find comprehensive information",
        llm=wrapped_llm,
    )

with agentreplay.AgentContext(agent_id="writer", workflow_id="article-creation"):
    writer = Agent(
        role="Content Writer",
        goal="Write engaging articles",
        llm=wrapped_llm,
    )

# Run the crew
crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])
result = crew.kickoff()

agentreplay.flush()

LlamaIndex

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
import agentreplay

agentreplay.init()

@agentreplay.traceable(name="index_documents", kind=agentreplay.SpanKind.EMBEDDING)
def build_index(directory: str):
    documents = SimpleDirectoryReader(directory).load_data()
    return VectorStoreIndex.from_documents(documents)

@agentreplay.traceable(name="query_index", kind=agentreplay.SpanKind.RETRIEVER)
def query(index, question: str):
    query_engine = index.as_query_engine()
    return query_engine.query(question)

index = build_index("./documents")
response = query(index, "What is the main topic?")
agentreplay.flush()

---

🌐 Environment Variables

Variable	Description	Default
AGENTREPLAY_API_KEY	API key for authentication	Required
AGENTREPLAY_PROJECT_ID	Project identifier	Required
AGENTREPLAY_BASE_URL	API base URL	https://api.agentreplay.io
AGENTREPLAY_TENANT_ID	Tenant identifier	default
AGENTREPLAY_AGENT_ID	Default agent ID	default
AGENTREPLAY_ENABLED	Enable/disable tracing	true
AGENTREPLAY_DEBUG	Enable debug logging	false
AGENTREPLAY_BATCH_SIZE	Spans per batch	100
AGENTREPLAY_FLUSH_INTERVAL	Auto-flush interval (seconds)	5.0
AGENTREPLAY_CAPTURE_INPUT	Capture function inputs	true
AGENTREPLAY_CAPTURE_OUTPUT	Capture function outputs	true

---

🧪 Testing

Disable Tracing in Tests

import agentreplay
import pytest

@pytest.fixture(autouse=True)
def disable_tracing():
    """Disable tracing for all tests."""
    agentreplay.init(enabled=False)
    yield
    agentreplay.reset()

def test_my_function():
    # Tracing is disabled, no network calls
    result = my_traced_function("test")
    assert result == expected

Or use environment variable:

AGENTREPLAY_ENABLED=false pytest

Mock the SDK

from unittest.mock import patch

def test_with_mock():
    with patch('agentreplay.flush'):
        # flush() won't actually send data
        result = my_function()

---

📚 Complete API Reference

Top-Level Functions

Function	Description
init(**config)	Initialize the SDK with configuration
flush(timeout=None)	Send all pending traces
shutdown(timeout=None)	Graceful shutdown with flush
reset()	Reset SDK state completely
get_stats()	Get diagnostic statistics
ping()	Check backend connectivity

Decorators & Tracing

Function	Description
@traceable	Decorator for function tracing
@observe	Alias for @traceable (Langfuse-style)
trace(name, **opts)	Context manager for creating spans
start_span(name, **opts)	Create a manual span
get_current_span()	Get the currently active span

Client Wrappers

Function	Description
wrap_openai(client, **opts)	Wrap OpenAI client
wrap_anthropic(client, **opts)	Wrap Anthropic client
wrap_method(obj, method, **opts)	Wrap any method

Context Management

Function	Description
set_context(**ctx)	Set global context
get_global_context()	Get current global context
clear_context()	Clear all global context
with_context(**ctx)	Scoped context manager
AgentContext(...)	Class-based agent context

Privacy

Function	Description
configure_privacy(**opts)	Configure redaction settings
redact_payload(data)	Redact sensitive data from dict
redact_string(text)	Redact patterns from string
hash_pii(value, salt=None)	Hash PII for anonymization
add_pattern(regex)	Add redaction pattern at runtime
add_scrub_path(path)	Add scrub path at runtime

ActiveSpan Methods

Method	Description
set_input(data)	Set span input data
set_output(data)	Set span output data
set_attribute(key, value)	Set a single attribute
set_attributes(dict)	Set multiple attributes
add_event(name, attrs)	Add a timestamped event
set_error(exception)	Record an error
set_token_usage(...)	Set LLM token counts
set_model(model, provider)	Set model information
end()	End the span

---

🤝 Contributing

We welcome contributions! See CONTRIBUTING.md for guidelines.

# Clone the repository
git clone https://github.com/agentreplay/agentreplay.git
cd agentreplay/sdks/python

# Create virtual environment
python -m venv .venv
source .venv/bin/activate  # or `.venv\Scripts\activate` on Windows

# Install in development mode
pip install -e ".[dev]"

# Run tests
pytest tests/ -v

# Run linter
ruff check src/

# Run type checker
mypy src/agentreplay

# Run formatter
ruff format src/

---

📄 License

Apache 2.0 - see LICENSE for details.

---

🔗 Links

• 📖 Documentation
• 💻 GitHub Repository
• 📦 PyPI Package
• 💬 Discord Community
• 🐦 Twitter

---

Made with ❤️ by the Agentreplay team