Metadata-Version: 2.4
Name: verse-sdk
Version: 0.3.0b6
Summary: Verse Python SDK
License: MIT
License-File: LICENSE
Requires-Python: >=3.12
Requires-Dist: cuid2>=2.0.1
Requires-Dist: opentelemetry-api>=1.34.1
Requires-Dist: opentelemetry-exporter-otlp[logs]>=1.34.1
Requires-Dist: opentelemetry-instrumentation-logging>=0.44b0
Requires-Dist: opentelemetry-instrumentation>=0.44b0
Requires-Dist: opentelemetry-sdk[logs]<1.35.0,>=1.34.1
Requires-Dist: requests>=2.31.0
Description-Content-Type: text/markdown

# Verse Python SDK

A Python SDK for observability and tracing in AI applications. Supports decorators and context managers with automatic instrumentation for popular LLM frameworks.

## Installation

```bash
pip install verse-sdk
```

## Quick Start

```python
from verse_sdk import verse, observe

# Initialize
verse.init(
    app_name="my-app",
    exporters=[verse.exporters.console()],
    vendors=["pydantic_ai"]  # Optional: auto-instrument LLM calls
)

# Option 1: Decorators (recommended)
@observe()
async def my_function(query: str):
    result = await process_query(query)
    return result

# Option 2: Context managers
async def my_function_v2(query: str):
    with verse.trace("my_function") as trace:
        trace.input(query)
        with verse.span("process_query") as span:
            result = await process_query(query)
            span.output(result)
        trace.output(result)
    return result
```

## Table of Contents

- [Initialization](#initialization)
- [Exporters](#exporters)
- [Decorators](#decorators)
- [Context Managers](#context-managers)
- [Context Methods](#context-methods)
- [Integrations](#integrations)
- [Examples](#examples)
- [API Reference](#api-reference)

## Initialization

```python
verse.init(
    app_name="my-app",           # Required: identifies your project
    environment="production",     # Optional: environment label
    exporters=[...],             # Required: list of exporters
    vendors=["pydantic_ai"],     # Optional: enables auto-instrumentation
    version="1.0.0"              # Optional: app version
)
```

## Exporters

### Console
```python
verse.exporters.console()
verse.exporters.console({"scopes": ["agent-workflow-1"]})  # With scope filtering
```

### Langfuse
```python
from verse_sdk import LangfuseConfig

verse.exporters.langfuse(
    LangfuseConfig(
        host="https://cloud.langfuse.com",
        public_key="pk-...",
        private_key="sk-..."
    )
)

# Or use environment variables: LANGFUSE_HOST, LANGFUSE_PUBLIC_KEY, LANGFUSE_PRIVATE_KEY
verse.exporters.langfuse()
```

### OTEL
```python
from verse_sdk import OtelConfig

verse.exporters.otel(
    OtelConfig(host="http://localhost:4318")
)
```

### Verse
```python
from verse_sdk import VerseConfig

verse.exporters.verse(
    VerseConfig(
        api_key="your-api-key",
        host="http://localhost:4318",
        project_id="your-project-id"
    )
)

# Or use environment variables: VERSE_API_KEY, VERSE_HOST, VERSE_PROJECT_ID
verse.exporters.verse()
```

### Scope Filtering

Route traces to specific exporters by scope:

```python
# Configure exporters with scopes
verse.init(
    app_name="my-app",
    exporters=[
        verse.exporters.console({"scopes": ["agent-a"]}),
        verse.exporters.langfuse({"scopes": ["agent-b"]})
    ]
)

# Set scope on traces
@observe(type="trace", scope="agent-a")
def my_function():
    pass
```

## Decorators

Decorators automatically capture inputs, outputs, and errors.

### Available Decorators

```python
from verse_sdk import observe
```

### Basic Usage

```python
@observe(type="trace")
async def answer_question(question: str):
    context = await retrieve_context(question)
    return await generate_answer(question, context)

@observe(type="span")
async def retrieve_context(question: str):
    return await db.search(question)

@observe(type="generation")
async def generate_answer(question: str, context: str):
    return await llm.complete(f"Context: {context}\nQ: {question}")

@observe(type="tool")
def search_database(query: str):
    return db.search(query)
```

### Customization

```python
# Custom name
@observe(name="custom_name")
def my_function():
    pass

# Disable input/output capture
@observe(capture_input=False, capture_output=False)
def sensitive_llm_call():
    pass

# Add custom attributes
@observe(level="debug", custom_attr="value")
def detailed_operation():
    pass
```

### Accessing Current Context

```python
from verse_sdk import get_current_trace_context, get_current_span_context

@observe()
def workflow():
    trace_ctx = get_current_trace_context()
    trace_ctx.user("user-123").session("session-456")
    return process_data()

@observe()
def process_data():
    span_ctx = get_current_span_context()
    span_ctx.level("info").metadata({"step": "processing"})
    return "result"
```

## Context Managers

Context managers provide fine-grained control over when attributes are set.

### Basic Usage

```python
def process_request(user_id: str, query: str):
    with verse.trace("process_request") as trace:
        trace.input({"user_id": user_id, "query": query})
        trace.session(user_id).user(user_id)

        with verse.span("validate") as span:
            span.input(query).level("debug")
            is_valid = validate(query)
            span.output(is_valid)

        if is_valid:
            with verse.generation("llm_call") as gen:
                gen.model("gpt-4").vendor("openai").input(query)
                response = llm.complete(query)
                gen.output(response).usage({
                    "input_tokens": 150,
                    "output_tokens": 50,
                    "total_tokens": 200
                })

        trace.output(response)
        return response
```

### Grouped Context Managers (Python 3.10+)

```python
with (
    verse.trace("my_trace", session_id="user-123") as trace,
    verse.span("my_span", level="info") as span,
):
    result = process()
    span.output(result)
    trace.output(result)
```

### Setting Attributes

```python
# Option 1: During initialization
with verse.trace(name="my_trace", session_id="user-123", scope="agent-a"):
    pass

# Option 2: After initialization (chainable)
with verse.trace("my_trace") as trace:
    trace.session("user-123").scope("agent-a").tags(["production"])
```

## Context Methods

### Common Methods (All Contexts)

- `.input(data)` - Set input
- `.output(data)` - Set output
- `.metadata(dict)` - Add metadata
- `.error(exception)` - Record error
- `.score(dict)` - Add evaluation score
- `.event(name, level, **attrs)` - Add event
- `.set_attributes(**kwargs)` - Set custom attributes

### TraceContext

- `.session(session_id)` - Set session ID
- `.user(user_id)` - Set user ID
- `.scope(scope)` - Set scope for filtering
- `.tags(list)` - Add tags

### SpanContext

- `.level(level)` - Set log level ("info", "debug", "warning")
- `.operation(op)` - Set operation type ("tool", "db.query")
- `.status_message(message)` - Set status message

### GenerationContext

Inherits SpanContext methods, plus:

- `.model(model_name)` - Set model identifier
- `.vendor(vendor)` - Set model vendor
- `.usage(dict)` - Set token usage (`{"input_tokens": 150, "output_tokens": 50, "total_tokens": 200}`)
- `.messages(list)` - Set message history

## Integrations

Enable auto-instrumentation by setting the `vendors` parameter:

### Pydantic AI
```python
verse.init(app_name="my-app", vendors=["pydantic_ai"], exporters=[...])

agent = Agent("openai:gpt-4")
result = await agent.run("query")  # Automatically traced

# Streaming is also supported
async for event in agent.run_stream("query"):  # Automatically traced
    print(event)
```

**Supported:**
- ✅ `Agent.run()` - Synchronous and asynchronous runs are automatically traced
- ✅ `Agent.run_stream()` - Streaming runs are automatically traced with delta events

### LiteLLM
```python
verse.init(app_name="my-app", vendors=["litellm"], exporters=[...])

from litellm import completion
response = completion(model="gpt-4", messages=[...])  # Automatically traced
```

**Supported:**
- ✅ `completion()` - Synchronous and asynchronous completions are automatically traced
- ✅ Streaming completions - Stream events are captured and traced
- ✅ Embeddings - Embedding operations are automatically traced

### LangChain
```python
verse.init(app_name="my-app", vendors=["langchain"], exporters=[...])
# Your LangChain code is automatically traced
```

**Supported:**
- ✅ Chat models - `ChatModel` invocations (synchronous and asynchronous) are automatically traced
- ✅ LLM completions - `LLM` invocations (synchronous and asynchronous) are automatically traced
- ✅ Streaming - Streaming token events are captured via `on_llm_new_token`
- ✅ Embeddings - Embedding operations are automatically traced

### Anthropic
```python
verse.init(app_name="my-app", vendors=["anthropic"], exporters=[...])

from anthropic import Anthropic
client = Anthropic()
response = client.messages.create(model="claude-3-5-sonnet-20241022", messages=[...])  # Automatically traced
```

**Supported:**
- ✅ `Messages.create()` - Synchronous message creation is automatically traced
- ✅ `AsyncMessages.create()` - Asynchronous message creation is automatically traced
- ✅ Streaming - Streaming responses are automatically traced with delta events

### Google (Gemini)
```python
verse.init(app_name="my-app", vendors=["google"], exporters=[...])

from google.genai import Client
client = Client()
response = client.models.generate_content(model="gemini-pro", contents=[...])  # Automatically traced
```

**Supported:**
- ✅ `Models.generate_content()` - Synchronous content generation is automatically traced
- ✅ `Models.generate_content_stream()` - Streaming content generation is automatically traced

## Agent Fixtures and Examples

The repository includes complete working examples of agents using each supported integration. These fixtures demonstrate best practices for setting up tools, handling streaming, and managing complex agent workflows.

### Available Agent Examples

**[Anthropic Agent](https://github.com/TextLayer/verse-sdk/blob/main/tests/fixtures/agents/anthropic_agent.py)**
- Tool calling with function execution
- Streaming responses with tool support
- Multi-turn conversations with tool results

**[Google (Gemini) Agent](https://github.com/TextLayer/verse-sdk/blob/main/tests/fixtures/agents/google_agent.py)**
- Function calling with Google GenAI
- Streaming content generation
- Tool execution and response handling

**[LangChain Agent](https://github.com/TextLayer/verse-sdk/blob/main/tests/fixtures/agents/langchain_agent.py)**
- ChatOpenAI with tool binding
- Streaming with tool calls
- Agent executor patterns

**[LiteLLM Agent](https://github.com/TextLayer/verse-sdk/blob/main/tests/fixtures/agents/litellm_agent.py)** and **[LiteLLM Utilities](https://github.com/TextLayer/verse-sdk/blob/main/tests/fixtures/agents/litellm.py)**
- Complete tool calling setup with LiteLLM
- Function schema generation
- Async completion with tools
- Tool execution loop implementation

**[Pydantic AI Agent](https://github.com/TextLayer/verse-sdk/blob/main/tests/fixtures/agents/pydantic_agent.py)**
- Agent with tool definitions
- Streaming with Pydantic AI
- Type-safe tool calling

### Using the Examples

These fixtures are fully functional and can be used as templates for your own agents:

```python
# Example: Using the LiteLLM agent fixture
from tests.fixtures.agents.litellm_agent import LiteLLMAgent

agent = LiteLLMAgent()
response = await agent.ask("What's the weather in San Francisco?")
```

Each agent fixture includes:
- Complete setup and initialization
- Tool/function definitions
- Streaming support
- Error handling
- Integration with Verse SDK tracing

## API Reference

### Decorators

**`@observe(name=None, type=str capture_input=True, capture_output=True, capture_metadata=True, observation_type=str, **attrs)`**

Creates an observation.

### Context Managers

**`verse.trace(name, session_id=None, user_id=None, scope=None, tags=None, metadata=None, **attrs)`**

Returns `TraceContext`

**`verse.span(name, input=None, output=None, level=None, op=None, status_message=None, metadata=None, **attrs)`**

Returns `SpanContext`

**`verse.generation(name, model=None, vendor=None, input=None, output=None, messages=None, usage=None, **attrs)`**

Returns `GenerationContext`

### Validation

**`OtelGenAISpecCheck(raw_span)`**

A specification checker that validates spans against the OpenTelemetry GenAI semantic conventions. This tool helps ensure your LLM tracing data is healthy, compliant, and complete.

#### Purpose

The `OtelGenAISpecCheck` class:
- **Validates required attributes** - Identifies missing required fields that could break downstream processing
- **Detects deprecated attributes** - Flags attributes that should be migrated to newer conventions
- **Checks attribute values** - Ensures values conform to allowed enums and formats
- **Highlights recommendations** - Surfaces optional attributes that improve trace quality
- **Filters by context** - Only validates attributes relevant to the specific provider and operation

#### Basic Usage

```python
from verse_sdk.spec import OtelGenAISpecCheck

# Validate a span
checker = OtelGenAISpecCheck(span_data)

# Check for issues
if checker.has_errors:
    print("Span has validation errors")

if checker.could_improve:
    print("Span is missing recommended attributes")

if checker.has_deprecations:
    print("Span uses deprecated attributes")
```

#### Validation Methods

**Checking Overall Status:**
- `has_errors` - Returns `True` if span has required attributes missing or invalid values
- `could_improve` - Returns `True` if span is missing recommended (but not required) attributes
- `has_deprecations` - Returns `True` if span uses deprecated attributes

**Checking Specific Attributes:**
- `is_attribute_valid(name)` - Check if a specific attribute is valid
- `is_attribute_invalid(name)` - Check if a specific attribute has errors
- `is_attribute_missing(name)` - Check if a specific attribute is missing
- `is_attribute_deprecated(name)` - Check if a specific attribute is deprecated

**Extracting Data:**
- `extract_metadata()` - Get structured metadata (model, session, trace IDs, etc.)
- `extract_session_id()` - Get session ID from standard or legacy attribute names
- `get_attribute_value(name, default)` - Get any span attribute value
- `get_span_context_value(name, default)` - Get span context fields (span_id, trace_id)
- `get_span_root_value(name, default)` - Get root-level span fields (start_time, end_time)

#### Example: Validating Exported Spans

```python
from verse_sdk import verse
from verse_sdk.spec import OtelGenAISpecCheck

# Custom exporter that validates spans
class ValidatingExporter:
    def export(self, spans):
        for span in spans:
            checker = OtelGenAISpecCheck(span)

            if checker.has_errors:
                print(f"ERROR: Span {checker.extract_metadata().span_id} has validation errors")

            if checker.could_improve:
                print(f"WARNING: Span could be improved with recommended attributes")

            if checker.is_attribute_deprecated("gen_ai.system"):
                print("INFO: Migrate 'gen_ai.system' to 'gen_ai.provider.name'")

verse.init(
    app_name="my-app",
    exporters=[ValidatingExporter()]
)
```

#### Example: Extracting Metadata

```python
from verse_sdk.spec import OtelGenAISpecCheck

checker = OtelGenAISpecCheck(span_data)
metadata = checker.extract_metadata()

print(f"Model: {metadata.model}")
print(f"Provider: {metadata.model_provider}")
print(f"Session: {metadata.session_id}")
print(f"Trace: {metadata.trace_id}")
print(f"Environment: {metadata.environment}")
```

### Helper Functions

**`get_current_trace_context() -> TraceContext`**

Get the current trace context from active span. Raises `ValueError` if unavailable.

**`get_current_span_context() -> SpanContext`**

Get the current span context from active span. Raises `ValueError` if unavailable.

**`get_current_generation_context() -> GenerationContext`**

Get the current generation context from active span. Raises `ValueError` if unavailable.

### Data Formats

**Usage:**
```python
{"input_tokens": 150, "output_tokens": 50, "total_tokens": 200}
```

**Score:**
```python
{"name": "quality", "value": 0.95, "comment": "Excellent"}
```

## Best Practices

1. **Use decorators by default** - Cleaner code with automatic input/output capture
2. **Use context managers for fine-grained control** - When you need dynamic names or conditional logic
3. **Combine both approaches** - Decorators for functions, context managers within functions
4. **Choose appropriate observation types**:
   - **Trace**: Top-level workflows
   - **Span**: Sub-operations, processing steps
   - **Generation**: LLM API calls
   - **Tool**: Function/tool calls in agent systems
5. **Enable auto-instrumentation** - Set `vendors` parameter for supported frameworks
6. **Use scope filtering** - Route traces to different exporters by scope

## Shutdown

```python
verse.shutdown()  # Flush all traces before exit
```

## Support

For issues and questions, please open an issue on GitHub.
