zeno-qdrant

Qdrant-backed KnowledgeStore adapter for Zeno.

Persists facts and documents in either a local on-disk Qdrant store (no server needed) or a remote Qdrant instance. Each row carries a user_id payload field so query-time filters enforce multi-tenant isolation at the database layer. Plugs into the ThreeLayer memory binder as the knowledge and (optionally) the user_memory backing.

Install

uv add 'zeno-framework[qdrant]'
# or, without the AI package:
uv add zeno-qdrant

Features

• Native async client (qdrant-client's AsyncQdrantClient) — no thread-pool wrapping
• Local file store via path= (single-process, embedded use cases) or remote server via url= and api_key=
• Bundled FastEmbed (BAAI/bge-small-en-v1.5, 384-dim) for the default embedding path; no external embedding service required
• user_id-scoped search via Qdrant's Filter + FieldCondition payload filter (S7 isolation enforced server-side)
• Lazy collection auto-creation on first add()

Minimal usage

import asyncio
from pathlib import Path

from zeno.agent import Agent
from zeno.app import ZenoApp
from zeno.builtin_tools import recall, remember
from zeno.channels.cli import CliChannel
from zeno.qdrant import QdrantKnowledgeStore
from zeno.memory.sqlite.conversation_store import SqliteConversationStore
from zeno.memory.sqlite.session_store import SqliteSessionStore
from zeno.memory.three_layer import ThreeLayer
from zeno.memory.vector_backed.user_memory_store import VectorBackedUserMemoryStore
from zeno.testing import FakeProvider

async def main() -> None:
    data_dir = Path("./.zeno")
    data_dir.mkdir(exist_ok=True)

    # Local on-disk store — no server required.
    knowledge = QdrantKnowledgeStore(path=data_dir / "qdrant")
    # Or talk to a remote cluster:
    # knowledge = QdrantKnowledgeStore(url="https://...", api_key="...")

    memory = ThreeLayer(
        session=SqliteSessionStore(data_dir / "sessions.db"),
        user_memory=VectorBackedUserMemoryStore(knowledge_store=knowledge),
        knowledge=knowledge,
        conversation=SqliteConversationStore(data_dir / "conversations.db"),
    )

    app = ZenoApp(
        agent=Agent(
            name="root",
            instructions="Use `recall` to look things up and `remember` to store facts.",
            tools=[recall, remember],
        ),
        channels=[CliChannel()],
        provider=FakeProvider(),
        memory=memory,
    )
    await app.run()

asyncio.run(main())

Pass exactly one of path= or url= — passing both (or neither) raises ValueError at construction so the failure surfaces before the first turn rather than mid-search.

Testing

uv run pytest packages/zeno-qdrant

Tests use a temporary local Qdrant directory. The contract suite shared with zeno-chroma guarantees both adapters honour the same KnowledgeStore semantics — swap one for the other with no test changes.

See also

• zeno-chroma — the other first-party KnowledgeStore adapter.
• docs/memory.md — three-layer memory model.
• docs/adapters.md — writing your own KnowledgeStore.

Part of the Zeno framework.