thinkhive 2.0.0

Python SDK for ThinkHive - AI Agent Observability Platform

ThinkHive Python SDK v2.0.0

The official Python SDK for ThinkHive - AI Agent Observability Platform.

Note: The Python SDK supports V1 APIs (tracing, analysis, quality metrics, analytics). For V3 APIs (runs, claims, calibration, ticket linking, configurable ROI), use the JavaScript/TypeScript SDK.

Installation

pip install thinkhive

For framework-specific auto-instrumentation:

pip install thinkhive[langchain]     # LangChain support
pip install thinkhive[llamaindex]    # LlamaIndex support
pip install thinkhive[openai]        # OpenAI support
pip install thinkhive[all]           # All frameworks

Quick Start

from thinkhive import ThinkHive

client = ThinkHive(api_key="your_api_key")

result = client.trace(
    user_message="What is the weather in San Francisco?",
    agent_response="The weather in San Francisco is currently 65F and sunny.",
    agent_id="weather-agent"
)

print(f"Trace ID: {result.trace_id}")
if result.analysis:
    print(f"Outcome: {result.analysis.outcome_verdict}")
    print(f"Impact Score: {result.analysis.impact_score}")

API Reference

Tracing

from thinkhive import ThinkHive, BusinessContext

client = ThinkHive(api_key="your_api_key")

# Basic trace
result = client.trace(
    user_message="I want to cancel my order #12345",
    agent_response="I understand you want to cancel order #12345...",
    agent_id="support-agent",
)

# With business context for ROI analysis
result = client.trace(
    user_message="I want to cancel my order #12345",
    agent_response="I understand you want to cancel order #12345...",
    agent_id="support-agent",
    business_context=BusinessContext(
        customer_id="cust_abc123",
        transaction_value=150.00,
        priority="high",
        industry="ecommerce"
    )
)

if result.analysis and result.analysis.roi_metrics:
    roi = result.analysis.roi_metrics
    print(f"Revenue Risk: ${roi.estimated_revenue_loss:.2f}")
    print(f"Churn Probability: {roi.churn_probability}%")

# Get an existing trace
trace = client.get_trace("trace_abc123")

# Request analysis for an existing trace
analysis = client.analyze_trace("trace_abc123")

Feedback

# Positive feedback
client.feedback(
    trace_id=result.trace_id,
    rating=5,
    was_helpful=True,
    comment="Very accurate response!"
)

# Negative feedback with correction
client.feedback(
    trace_id=result.trace_id,
    rating=2,
    was_helpful=False,
    had_issues=["incorrect_info", "too_long"],
    corrected_response="The correct answer is..."
)

Decorators and Context Managers

from thinkhive import ThinkHive, Tracer, trace_function

client = ThinkHive(api_key="your_api_key")

# Decorator-based tracing
@trace_function("my-agent", client)
def process_query(question: str) -> str:
    response = call_llm(question)
    return response

answer = process_query("What are your business hours?")

# Context manager with spans
tracer = Tracer(client)

with tracer.trace("my-agent") as ctx:
    ctx.user_message = "Find information about Python decorators"

    with tracer.span("retrieval", "retriever"):
        docs = vector_store.similarity_search(ctx.user_message)
        ctx.retrieved_contexts.extend([doc.page_content for doc in docs])

    with tracer.span("generation", "llm"):
        ctx.agent_response = llm.generate(ctx.user_message, docs)

Explainer API

# Analyze a trace with explainability
result = client.explainer.analyze(
    user_message="I need help with a refund",
    agent_response="I can process your refund...",
    outcome="success",
    business_context={"transactionValue": 150},
)

print(f"Summary: {result['summary']}")
print(f"Recommendations: {result['recommendations']}")

# Batch analysis
results = client.explainer.analyze_batch(
    traces=[
        {"userMessage": "msg1", "agentResponse": "resp1"},
        {"userMessage": "msg2", "agentResponse": "resp2"},
    ],
    tier="standard",
)

# Semantic search across traces
results = client.explainer.search(
    query="refund issues",
    filters={"outcome": "failure"},
    limit=10,
)

Quality Metrics API

# RAG quality scores
scores = client.quality.get_rag_scores(trace_id)
print(f"Groundedness: {scores['groundedness']}")
print(f"Faithfulness: {scores['faithfulness']}")

# Hallucination detection
report = client.quality.get_hallucination_report(trace_id)
if report['hasHallucinations']:
    for detection in report['detectedTypes']:
        print(f"  - {detection['type']}: {detection['description']}")

# Check RAG quality for custom input
check = client.quality.evaluate_rag(
    query="What is the return policy?",
    response="Items can be returned within 30 days.",
    contexts=[{"content": "Return Policy: 30 day returns..."}],
)

# Detect hallucinations
detection = client.quality.detect_hallucinations(
    response="The product weighs 5 kg.",
    contexts=[{"content": "Product weight: 3 kg"}],
)

Analytics API

# ROI summary
summary = client.analytics.get_roi_summary()
print(f"Revenue Saved: ${summary['totalRevenueSaved']:,.2f}")

# Per-agent ROI
agent_roi = client.analytics.get_roi_by_agent("support-agent")
print(f"Success Rate: {agent_roi['successRate']}%")

# Trends
trends = client.analytics.get_trends(period="30d", agent_id="support-agent")

# Correlation analysis
correlations = client.analytics.get_correlations()
for corr in correlations['correlations']:
    print(f"{corr['type']}: {corr['actionableInsight']}")

# Calculate ROI for traces
roi = client.analytics.calculate_roi(
    traces=[{"userMessage": "...", "agentResponse": "..."}],
    industry="saas",
)

Auto-Instrumentation

from thinkhive import ThinkHive, auto_instrument, BusinessContext

client = ThinkHive(api_key="your_api_key")

auto_instrument(
    client,
    frameworks=["langchain", "openai"],
    capture_prompts=True,
    capture_responses=True,
    business_context=BusinessContext(industry="saas")
)

# All LangChain and OpenAI calls are now automatically traced
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4")
response = llm.invoke("What is machine learning?")  # Automatically traced

Async Mode

client = ThinkHive(
    api_key="your_api_key",
    async_mode=True  # Traces are sent asynchronously in batches
)

for query, response in interactions:
    client.trace(
        user_message=query,
        agent_response=response,
        agent_id="batch-agent"
    )

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

Environment Variables

Variable	Description
THINKHIVE_API_KEY	Your ThinkHive API key
THINKHIVE_BASE_URL	Custom API endpoint (default: https://api.thinkhive.ai)

V3 API Support

The Python SDK currently supports V1 APIs. For V3 features, use the JavaScript SDK:

Feature	Python SDK	JS SDK
Tracing (V1)	client.trace()	traceLLM(), traceRetrieval()
Explainer (V1)	client.explainer	analyzer
Quality (V1)	client.quality	qualityMetrics
Analytics (V1)	client.analytics	roiAnalytics (V1 methods)
Runs (V3)	--	runs
Claims (V3)	--	claims
Calibration (V3)	--	calibration
Ticket Linking (V3)	--	linking
ROI Config (V3)	--	roiAnalytics (V3 methods)
Customer Context (V3)	--	customerContext

License

MIT License - see LICENSE for details.