Python SDK

The Seer Python SDK provides a lightweight client for logging retrieval events.

Beta Access Required

The SDK is distributed to beta users directly. Email ben@seersearch.com to request access and receive installation instructions.

Installation

Once you have beta access, install the SDK:

pip install seer-sdk

# With OpenTelemetry support
pip install seer-sdk[otel]

Quick Start

from seer import SeerClient

client = SeerClient()  # reads SEER_API_KEY from env

client.log(
    task="How do I reset my password?",
    context=[
        {"text": "To reset your password, go to Settings...", "score": 0.89}
    ],
)
# Events auto-flush on process exit

---

Configuration

Environment Variables

Variable	Description	Default
SEER_API_KEY	API key (required)	—

Constructor Options

client = SeerClient(
    api_key="seer_live_...",           # or from SEER_API_KEY env
    fire_and_forget=True,              # async (default) or sync
    max_queue_size=10_000,             # max events before dropping
    flush_interval=0.5,                # worker flush interval (seconds)
    timeout=5.0,                       # HTTP timeout
)

---

SeerClient API

log()

Log a retrieval event.

client.log(
    # Required
    task: str,                                # The user query
    context: list[dict | str],                # Retrieved passages

    # Metadata
    metadata: dict | None = None,             # Free-form metadata

    # OpenTelemetry (auto-detected by default)
    trace_id: str | None = None,              # OTEL trace ID (32 hex)
    span_id: str | None = None,               # OTEL span ID (16 hex)
    parent_span_id: str | None = None,        # OTEL parent span ID
    span_name: str | None = None,             # Operation type
    use_otel_trace: bool = True,              # Auto-detect OTEL context

    # Multi-hop / Agentic
    is_final_context: bool = False,           # Mark as final evidence
    subquery: str | None = None,              # Decomposed sub-question

    # Accuracy testing
    ground_truth: dict | None = None,         # For comparing against expected

    # Other options
    created_at: str | None = None,            # ISO8601 timestamp override
    sample_rate: float | None = None,         # 0.0-1.0 sampling rate
) -> str | None

Returns:

• fire_and_forget=True (default): None (event queued async)
• fire_and_forget=False: record_id string from API

flush()

Wait for all queued events to be sent.

client.flush(timeout: float | None = None)

stats()

Get client statistics.

stats = client.stats()
print(f"Sent: {stats.events_sent}, Failed: {stats.events_failed}")

Returns a ClientStats object:

@dataclass
class ClientStats:
    events_enqueued: int = 0
    events_sent: int = 0
    events_dropped: int = 0
    events_failed: int = 0
    bytes_sent: int = 0
    last_error: str | None = None

close()

Shutdown client gracefully (flushes remaining events).

client.close()

---

Context Format

Passages can be simple strings or objects with metadata:

# Simple strings
context = ["Passage one...", "Passage two..."]

# Passage objects (recommended)
context = [
    {
        "text": "The main content...",    # Required
        "score": 0.95,                     # Optional: retrieval score
        "id": "doc-123",                   # Optional: document ID
        "source": "wiki",                  # Optional: source name
        "metadata": {"author": "..."},     # Optional: custom metadata
    }
]

---

Decorator

Use @seer_trace to automatically log function calls:

from seer import seer_trace

@seer_trace(
    task_arg="query",           # which argument is the query
    context_from_return=True,   # return value is the context
    metadata={"service": "help-bot"},
    sample_rate=0.10,
)
def retrieve(query: str) -> list[dict]:
    return [{"text": "Result...", "score": 0.9}]

# Automatically logged when called
results = retrieve("How do I reset my password?")

Decorator Parameters

Parameter	Type	Description
task_arg	str	Name of argument containing the query
context_from_return	bool	Use return value as context
metadata	dict	Static metadata to attach
sample_rate	float	Sampling rate (0.0-1.0)

---

Fire-and-Forget vs Synchronous

Fire-and-Forget (Default)

Events are queued and sent asynchronously. log() returns immediately.

client = SeerClient()  # fire_and_forget=True
client.log(task="...", context=[...])  # returns None, queued
# Auto-flushes on process exit

Note: Auto-flush via atexit happens on normal exit. Call flush() explicitly before os._exit() or in process pools.

Synchronous

Events are sent immediately. log() blocks and returns record_id.

client = SeerClient(fire_and_forget=False)
record_id = client.log(task="...", context=[...])
print(f"Created: {record_id}")

---

OpenTelemetry Integration

The SDK automatically captures OTEL trace context when available:

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span("retrieval"):
    # trace_id, span_id, span_name are captured automatically
    client.log(task="...", context=[...])

Manual Trace IDs

client.log(
    task="...",
    context=[...],
    trace_id="0af7651916cd43dd8448eb211c80319c",
    span_id="b7ad6b7169203331",
    span_name="retrieval",
    use_otel_trace=False,  # disable auto-detection
)

Span Name Patterns

Span names are used for filtering in the UI:

Pattern	UI Color
retrieval, retrieval_hop_N	Blue
rerank, reranker	Purple
llm_call, llm	Amber
synthesis, answer	Emerald

---

Ground Truth (Accuracy Testing)

Include expected results for accuracy measurement:

client.log(
    task="What is machine learning?",
    context=[
        {"text": "ML is a subset of AI...", "id": "doc-ml-intro"},
        {"text": "Neural networks learn from data...", "id": "doc-nn-basics"},
    ],
    ground_truth={
        # Document IDs that are relevant (matched against passage.id)
        "gold_doc_ids": ["doc-ml-intro", "doc-nn-basics"],
        "answer": "ML is a type of AI that learns from data",  # optional
    },
)

---

Convenience Function

For simple cases, use the global seer_log:

from seer import seer_log

seer_log(
    task="Quick question",
    context=[{"text": "Answer..."}],
)

Creates a global SeerClient on first call.

---

Error Handling

The SDK is designed to be non-blocking and fail gracefully:

• Queue overflow: Events are dropped (logged to seer.client logger)
• Network errors: Retried with exponential backoff, then dropped
• Invalid API key: Warning logged, event sent (server rejects)

Check client.stats() for failure counts.

---

Examples

See the SDK examples directory for complete usage patterns:

• Basic usage — All common patterns
• OpenTelemetry integration — Distributed tracing
• Replay rollouts — Batch evaluation from datasets