Python SDK

pip install intellirepo

from intellirepo import IntelliRepo
 
client = IntelliRepo(api_key="rh_live_xxx")
 
# Ask a question
response = client.chat.ask(question="What is the refund policy?")
 
print(response.answer)
print("Sources:", response.sources)

client = IntelliRepo(
    # Required
    api_key="rh_live_xxx",
 
    # Optional: Custom API URL
    base_url="https://api.intellirepo.ai",
 
    # Optional: Request timeout (seconds)
    timeout=30.0
)

with IntelliRepo(api_key="rh_live_xxx") as client:
    results = client.search.query(query="test")
# Client is automatically closed

result = client.collections.list()
 
for collection in result.collections:
    print(collection.name, collection.document_count)

collection = client.collections.create(
    name="Product Docs",
    description="All product documentation"
)

collection = client.collections.get("collection-uuid")

updated = client.collections.update("collection-uuid", name="New Name")

client.collections.delete("collection-uuid")

result = client.documents.list(collection_id)
 
for doc in result.documents:
    print(doc.name, doc.status)

result = client.documents.upload(
    collection_id,
    "path/to/document.pdf"
)

with open("document.pdf", "rb") as f:
    result = client.documents.upload(
        collection_id,
        f,
        filename="document.pdf"
    )

updated = client.documents.update(
    collection_id,
    document_id,
    tags=["important", "2024"]
)

client.documents.delete(collection_id, document_id)

response = client.chat.ask(question="What is the refund policy?")
print(response.answer)

response = client.chat.ask(
    question="What is the refund policy?",
    collection_id="collection-uuid"
)

response = client.chat.ask(
    question="What is the refund policy?",
    collection_id="collection-uuid",
    top_k=5,  # Number of sources to retrieve
    tags=["current"]  # Filter by document tags
)

for event in client.chat.stream(question="Explain the process..."):
    if event.type == "sources":
        print(f"Found {len(event.sources)} sources")
    elif event.type == "chunk":
        print(event.text, end="", flush=True)
    elif event.type == "done":
        print(f"\nTokens: {event.usage.input_tokens + event.usage.output_tokens}")
    elif event.type == "error":
        print(f"Error: {event.message}")

results = client.search.query(query="password reset")
 
for result in results.results:
    print(result.document_name, result.similarity)

results = client.search.query(
    query="password reset",
    collection_id="collection-uuid",
    top_k=10
)

from intellirepo import (
    IntelliRepoError,
    AuthenticationError,
    NotFoundError,
    RateLimitError,
    ValidationError
)
 
try:
    client.collections.get("invalid-id")
except AuthenticationError:
    print("Invalid API key")
except NotFoundError:
    print("Collection not found")
except RateLimitError:
    print("Rate limit exceeded")
except IntelliRepoError as e:
    print(f"API error: {e.message}")

import asyncio
from intellirepo import AsyncIntelliRepo
 
async def main():
    client = AsyncIntelliRepo(api_key="rh_live_xxx")
 
    response = await client.chat.ask(question="What is the refund policy?")
    print(response.answer)
 
    await client.close()
 
asyncio.run(main())

async with AsyncIntelliRepo(api_key="rh_live_xxx") as client:
    response = await client.chat.ask(question="What is the refund policy?")

from intellirepo.types import Collection, Document, ChatResponse
 
def process_response(response: ChatResponse) -> None:
    print(response.answer)

from flask import Flask, request, jsonify
from intellirepo import IntelliRepo
 
app = Flask(__name__)
client = IntelliRepo(api_key="rh_live_xxx")
 
@app.route("/ask", methods=["POST"])
def ask():
    question = request.json.get("question")
    response = client.chat.ask(question=question)
    return jsonify({
        "answer": response.answer,
        "sources": [s.document_name for s in response.sources]
    })