animuz-core

Core shared utilities for Animuz RAG (Retrieval-Augmented Generation) system.

Features

• Unified RAG API: 3-knob interface (prompt, llm, tools) for chat, ingestion, and retrieval
• LLM Clients: OpenAI (Responses API + Chat Completions), Anthropic Claude, Ollama
• Tool System: @tool decorator, qdrant_retriever() factory, MCP server support
• RAG Pipelines: Simple and Agentic RAG implementations (lower-level building blocks)
• Vector Database: Qdrant integration with hybrid search (dense + sparse)
• Embedding Clients: Multiple providers (local server, Modal, S3/SageMaker)
• Document Ingestion: Azure Document Intelligence, Unstructured, PDF extraction, structured text parsing
• CloudWatch Logging: Structured JSON logging with watchtower

Requirements

• Python >= 3.10

Installation

Install the core package (minimal dependencies only):

pip install animuz-core

Then install only the extras you need:

# Single extra
pip install animuz-core[openai]

# Multiple extras
pip install animuz-core[openai,qdrant,aws]

# Everything
pip install animuz-core[all]

Works with uv too:

uv pip install animuz-core[openai,qdrant]

Available Extras

The opt-in live V31MCPTools signed routes need boto3 at execution time. Install the aws extra (for example, animuz-core[openai,aws]) in a serving adapter that uses them. Fixture-only offline turns do not need AWS credentials. For a future ChatV4 live binding, construct V31MCPTools with an explicit, nonblank mcp_api_key and V31CatalogOptions(timeout_seconds=..., post=...). That opt-in rejects ambient-key fallback and gives /list_tools, legacy tool POSTs, and signed built-in POSTs a bounded connect/read timeout; it is not an overall wall-clock deadline. Transport, HTTP and malformed catalog-envelope or basic tool-shape failures raise a static CatalogUnavailable before a tool catalog is installed. Nested JSON Schema semantics are not fully validated here; a catalog can still fail at the model request. Omitting the option preserves current V31/legacy construction behavior. Strict tool execution makes one POST and returns a static no-guess result for transport, HTTP, malformed JSON, or nested failure envelopes. A timeout may occur after an action commits; this option does not provide exactly-once execution. Side-effecting tools still need a live fencing/receipt design before ChatV4 serves traffic. Strict mode also validates signed dispatcher failure envelopes before they reach the model; dispatcher calls retain their existing 8-second connect/read timeout. The opt-in requires function tool schemas; inventory the actual MCP catalog and fix any older definitions before enabling it in a serving adapter.

Usage

Unified RAG API (recommended)

The RAG class is the main entry point. It has 3 knobs:

• prompt — callable (team_id, assistant_id) -> dict that fetches an assistant config
• llm — model name string (provider auto-detected: "gpt-*" → OpenAI, "claude-*" → Anthropic)
• tools — list[ToolSpec] for local tools, MCP(url=...) for MCP server, or None for plain chat

from animuz_core import RAG, qdrant_retriever, tool

# Define how to fetch the assistant config
def my_fetcher(team_id, assistant_id):
    return {"prompt": "You are a helpful assistant.", "model": "gpt-4o-mini"}

# Create RAG with local retriever tool
rag = RAG(
    prompt=my_fetcher,
    llm="gpt-4o-mini",
    tools=[
        qdrant_retriever(host="localhost", port=6333, collection="animuz"),
    ],
)

# Chat (returns frontend-ready output)
output = await rag.chat("team1", "asst1", [{"role": "user", "content": "Hello"}])

# Ingest a document
await rag.add_doc("docs/intro.md", user_chat_id="team1|asst1")

# Retrieve documents
texts, points = await rag.retrieve("what is this?", user_chat_id="team1|asst1")

With MCP tools (Lambda / cloud)

from animuz_core import RAG, MCP

rag = RAG(
    prompt=ddb_fetcher,
    llm="gpt-4o-mini",
    tools=MCP(url=os.environ["MCP_URL"], api_key=os.environ.get("MCP_API_KEY")),
)
output = await rag.chat(team_id, assistant_id, messages, user_context=ctx)

Plain chat (no tools)

rag = RAG(prompt=my_fetcher, llm="gpt-4o-mini")
output = await rag.chat(team_id, assistant_id, messages)

Custom OpenAI base URL (proxy / gateway)

Route OpenAI API calls through a proxy or API gateway by passing base_url:

rag = RAG(
    prompt=my_fetcher,
    llm="gpt-4o-mini",
    base_url="https://your-proxy.example.com/openai/v1",
    tools=[qdrant_retriever(...)],
)

When omitted, the default OpenAI endpoint (api.openai.com) is used.

Custom tools with @tool decorator

from animuz_core import RAG, tool, qdrant_retriever

@tool(description="Get weather for a city")
async def weather(city: str) -> str:
    return await fetch_weather(city)

rag = RAG(
    prompt=my_fetcher,
    llm="gpt-4o-mini",
    tools=[qdrant_retriever(...), weather],
)

Lower-level APIs

LLM Clients

from animuz_core import OpenAIAgentClientResponses

# OpenAI Responses API agent with tool loop (recommended for production)
agent = OpenAIAgentClientResponses(user_chat_id="tenant-123", tools=tool_dict, model="gpt-4o-mini")
result = await agent.get_reply_frontend(messages, system_prompt)

# With a custom base URL (proxy / gateway)
agent = OpenAIAgentClientResponses(model="gpt-4o-mini", base_url="https://your-proxy.example.com/openai/v1")

RAG Pipelines

from animuz_core import SimpleRAG

# Simple RAG — always retrieves then generates
pipeline = SimpleRAG(
    embedding_client=embedding_client,
    db_client=qdrant_client,
    LLM=llm_client,
)
await pipeline.add_doc("document.pdf", user_chat_id="tenant-123")
result = await pipeline.query("What is RAG?", user_chat_id="tenant-123")

Vector Database

from animuz_core import QdrantDBClient

client = QdrantDBClient(host="localhost", port=6333, collection_name="animuz")
results = await client.hybrid_search(dense_vec, indices, values, user_chat_id="tenant-123")

Embedding

from animuz_core import EmbeddingClient, ModalEmbeddingClient

# Local embedding server
client = EmbeddingClient(host="localhost", port=12081)
result = await client.get_embedding("Some text")

# Modal-hosted embeddings
client = ModalEmbeddingClient()
result = client.get_embedding("Some text")

Development

# Clone and install in editable mode with dev dependencies
git clone <repo-url>
cd animuz-core
pip install -e ".[dev]"

# Run tests
pytest tests/

# Run integration tests (requires external services + env vars)
pytest -m integration tests/integration/
pytest -m integration tests/integration/test_e2e_rag_wrapper_simple.py

# Format
black src/
ruff check src/

Rudi ChatV4 turn coordinator

animuz_core.rudi_v4.run_turn is the shared async event generator for a future ChatV4 serving adapter and offline evaluation. It assembles the opt-in V31 prompt, maps messages, and runs the streaming agent. The caller must supply an authenticated assistant snapshot or resolver, state coordinator, tool gateway, model client, model settings, and clock. The gateway owns action and unknown-tool policy. This package does not provide a replay endpoint or a customer-data reader.

The generator emits one done event only after the final assistant message and all tool calls have matching outputs. Core response.completed events mark model rounds; they do not complete the turn. A caller must still implement delivery, authorization, and durable state for its own serving environment.

run_turn opts into V31-shaped metering: each completed tool result carries its reported per-call usage, and a final usage event follows the last assistant message before done. The final total includes outer model and reported tool tokens; unmeteredTools names tools whose costs are missing, so that total is a lower bound. Direct streaming-agent callers keep their previous event sequence unless they set emit_usage=True. Custom agent_factory implementations passed to run_turn must accept this keyword argument.

Injected gateways keep an eight-call default limit. A serving adapter can set ModelSettings(unbounded_tool_calls=True) to match V31's uncapped loop, after choosing its own request deadline and cancellation policy. Direct agent callers can use unbounded_tool_calls=True; it cannot be combined with max_tool_calls.

Publishing to PyPI

1. Bump the version in pyproject.toml and __init__.py.

2. Build the package:

uv pip install --upgrade build
uv run python -m build

3. (Optional) Verify the artifacts:

uv pip install --upgrade twine
uv run python -m twine check dist/*

4. Upload to TestPyPI first:

uv run python -m twine upload -r testpypi dist/*

5. Upload to PyPI:

uv run python -m twine upload dist/*

Notes:

• Create a PyPI API token and set TWINE_USERNAME=__token__ and TWINE_PASSWORD=<your-token>.
• If you upload to TestPyPI, install with pip install -i https://test.pypi.org/simple animuz-core to verify.

Integration Test Setup (Qdrant)

Use Docker Compose to run Qdrant locally:

docker compose -f docker-compose-qdrant.yml up -d qdrant

Then set the Qdrant env vars (example):

export QDRANT_HOST=localhost
export QDRANT_PORT=6333

Environment Variables

The package reads configuration from environment variables (loaded via python-dotenv):

License

MIT