lakehouse42-sdk 0.1.0

Python SDK for the Lakehouse42 API - Build RAG applications with ease

Lakehouse42 Python SDK

The official Python SDK for the Lakehouse42 API. Build powerful RAG (Retrieval-Augmented Generation) applications with ease.

Installation

pip install lakehouse42

Quick Start

import asyncio
from lakehouse42 import LakehouseClient

async def main():
    # Initialize the client
    async with LakehouseClient(api_key="lh_your_api_key") as client:
        # Create a document
        doc = await client.documents.create(
            title="Company Policies",
            content="""
            # Remote Work Policy

            Employees may work remotely up to 3 days per week.
            All remote work must be approved by your manager.
            """,
            content_type="text/markdown",
            metadata={"category": "HR", "department": "all"}
        )
        print(f"Created document: {doc.id}")

        # Search your knowledge base
        results = await client.search.query(
            "What is the remote work policy?",
            top_k=5
        )
        for result in results.results:
            print(f"- {result.document_title}: {result.score:.2f}")

        # Chat with RAG
        response = await client.chat.completions.create(
            messages=[
                {"role": "user", "content": "Summarize our remote work policy"}
            ],
            use_knowledge_base=True,
            include_sources=True
        )

        print(f"\nAssistant: {response.choices[0].message.content}")

        if response.sources:
            print("\nSources:")
            for source in response.sources:
                print(f"- {source.document_title}")

asyncio.run(main())

Synchronous Usage

If you prefer synchronous code, use LakehouseClientSync:

from lakehouse42 import LakehouseClientSync

client = LakehouseClientSync(api_key="lh_your_api_key")

# Create a document
doc = client.documents.create(
    title="Meeting Notes",
    content="Discussed Q1 roadmap priorities..."
)

# Search
results = client.search.query("Q1 roadmap")

# Chat
response = client.chat.completions.create(
    messages=[{"role": "user", "content": "What were the Q1 priorities?"}]
)

client.close()

Features

Documents

Create, upload, retrieve, and manage documents in your knowledge base.

# Create from text
doc = await client.documents.create(
    title="My Document",
    content="Document content...",
    metadata={"author": "John", "year": 2024}
)

# Upload a file
doc = await client.documents.upload(
    "/path/to/document.pdf",
    title="Uploaded Document"
)

# Get a document
doc = await client.documents.get("doc_123")

# List documents
result = await client.documents.list(limit=20)
for doc in result.data:
    print(f"{doc.title} - {doc.status}")

# Paginate through all documents
while result.has_more:
    result = await client.documents.list(cursor=result.next_cursor)

# Update metadata
doc = await client.documents.update(
    "doc_123",
    title="New Title",
    metadata={"reviewed": True}
)

# Delete
await client.documents.delete("doc_123")

Search

Search your knowledge base with semantic, keyword, or hybrid search.

# Basic hybrid search (recommended)
results = await client.search.query("machine learning concepts")

# Semantic search only
results = await client.search.query(
    "machine learning concepts",
    mode="vector"
)

# Keyword search only
results = await client.search.query(
    "machine learning",
    mode="keyword"
)

# With filters
results = await client.search.query(
    "quarterly revenue",
    filters=[
        {"field": "metadata.year", "operator": "gte", "value": 2023},
        {"field": "metadata.type", "operator": "eq", "value": "report"}
    ],
    top_k=10,
    rerank=True  # Apply cross-encoder reranking
)

# Find similar content
similar = await client.search.find_similar("chunk_456", top_k=5)

Chat Completions

Chat with your knowledge base using RAG.

# Simple chat with RAG
response = await client.chat.completions.create(
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What are our company policies on PTO?"}
    ],
    use_knowledge_base=True,
    include_sources=True
)

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

# Access sources
for source in response.sources:
    print(f"Source: {source.document_title} (relevance: {source.relevance_score:.2f})")

# Streaming
async for chunk in client.chat.completions.create_stream(
    messages=[{"role": "user", "content": "Explain quantum computing"}]
):
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
print()

# With session memory
response = await client.chat.completions.create(
    messages=[{"role": "user", "content": "Tell me about project X"}],
    session_id="session_123"  # Maintains conversation history
)

Collections

Organize documents into collections.

# Create a collection
collection = await client.collections.create(
    name="Engineering Docs",
    description="Technical documentation and runbooks",
    metadata={"team": "platform"}
)

# Add document to collection
doc = await client.documents.create(
    title="API Documentation",
    content="...",
    collection_id=collection.id
)

# Or add existing document
await client.collections.add_document(collection.id, "doc_123")

# List documents in collection
docs = await client.collections.list_documents(collection.id)

# Search within collection
results = await client.search.query(
    "deployment guide",
    collection_ids=[collection.id]
)

# Delete collection
await client.collections.delete(collection.id)

Streaming Search

Stream search results and generated answers in real-time.

sources = []
full_response = ""

async for event in client.search.stream("What is our PTO policy?"):
    if event.type == "sources":
        sources = event.sources
        print(f"Found {len(sources)} sources")
        for s in sources:
            print(f"  - {s.document_name}: {s.score:.2f}")
    elif event.type == "content":
        print(event.content, end="", flush=True)
        full_response += event.content or ""
    elif event.type == "metadata":
        print(f"\n\nTokens: {event.tokens_used}, Latency: {event.latency_ms}ms")
    elif event.type == "complete":
        print(f"Query ID: {event.query_id}")
    elif event.type == "error":
        print(f"Error: {event.error}")

Knowledge Graph / Entities

Query the knowledge graph of entities extracted from your documents.

# Get the knowledge graph
graph = await client.entities.get_graph(limit=200)
print(f"Nodes: {graph.stats.total_nodes}")
print(f"Edges: {graph.stats.total_edges}")

# Visualize nodes
for node in graph.nodes:
    print(f"- {node.name} ({node.type})")

# Get graph for a specific document
doc_graph = await client.entities.get_graph(document_id="doc_123")

# Search entities
entities = await client.entities.search(
    "John Smith",
    entity_types=["person"]
)

# Get related entities
related = await client.entities.get_related(
    entity.id,
    relationship_types=["works_for", "reports_to"]
)
for rel in related:
    print(f"{rel.source} --{rel.relationship_type}--> {rel.target}")

Error Handling

The SDK provides specific exception types for different error conditions:

from lakehouse42 import (
    LakehouseClient,
    LakehouseError,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
    ValidationError,
    QuotaExceededError,
    ConflictError,
)

async with LakehouseClient(api_key="lh_xxx") as client:
    try:
        doc = await client.documents.get("doc_123")
    except AuthenticationError as e:
        print(f"Invalid API key: {e.message}")
    except NotFoundError as e:
        print(f"Document not found: {e.message}")
    except RateLimitError as e:
        print(f"Rate limited. Retry after {e.retry_after} seconds")
    except QuotaExceededError as e:
        print(f"Quota exceeded: {e.message} (type: {e.quota_type})")
    except ConflictError as e:
        print(f"Conflict: {e.message}")
    except ValidationError as e:
        print(f"Invalid request: {e.message} (param: {e.param})")
    except LakehouseError as e:
        print(f"API error: {e.message}")

Retry Logic

The SDK includes automatic retry with exponential backoff for:

• Rate limit errors (429)
• Server errors (500, 502, 503, 504)
• Connection errors

Configure retry behavior:

client = LakehouseClient(
    api_key="lh_xxx",
    max_retries=3,  # Number of retries (default: 3)
    timeout=60.0,   # Request timeout in seconds (default: 60)
)

The SDK respects Retry-After headers when present.

Configuration

Client Options

client = LakehouseClient(
    api_key="lh_xxx",
    base_url="https://api.lakehouse42.com/v1",  # Custom API endpoint
    timeout=60.0,  # Request timeout in seconds
    max_retries=2,  # Number of retries for failed requests
)

Environment Variables

You can also set your API key via environment variable:

export LAKEHOUSE_API_KEY=lh_your_api_key

import os
from lakehouse42 import LakehouseClient

client = LakehouseClient(api_key=os.environ["LAKEHOUSE_API_KEY"])

Type Safety

The SDK is fully typed and works great with type checkers like mypy:

from lakehouse42 import LakehouseClient, Document, SearchResults

async def process_documents(client: LakehouseClient) -> list[Document]:
    result = await client.documents.list(limit=100)
    return result.data

async def search_and_process(client: LakehouseClient, query: str) -> SearchResults:
    return await client.search.query(query, top_k=10)

Requirements

• Python 3.8+
• httpx >= 0.25.0
• pydantic >= 2.0.0

License

MIT License - see LICENSE for details.

Links

• Documentation
• API Reference
• GitHub
• PyPI