Skip to content

symfonic.knowledge.adapters

adapters

Concrete KnowledgeProvider adapters.

ChromaKnowledgeProvider requires the knowledge extra (chromadb). It used to be imported here inside a try/except ImportError that appended it to __all__ on success -- which meant every import of this package probed for chromadb, and the package's export surface silently depended on what happened to be installed.

Both problems have the same fix: resolve the adapter through a module __getattr__ (PEP 562). The name is always exported, absence is reported at first use with the install command, and importing the package reaches no distribution at all (rule IA-5 of the integration-adapter standard, T4.2.1).

ChromaKnowledgeProvider 

ChromaKnowledgeProvider(path: str, collection_name: str)

KnowledgeProvider backed by a local ChromaDB persistent collection.

Connects to an existing ChromaDB directory and queries a named collection using ChromaDB's built-in default embedding function. For custom embeddings, pre-embed queries externally and use a different adapter.

All synchronous ChromaDB operations are wrapped with asyncio.to_thread.

Parameters:

Name Type Description Default
path str

Filesystem path to the ChromaDB persistent directory.

 required
collection_name str

Name of the collection to query.

 required

Example::

provider = ChromaKnowledgeProvider(
    path="/data/chroma",
    collection_name="legal_docs",
)
fragments = await provider.search("force majeure clause", limit=5)
Source code in symfonic/knowledge/adapters/chroma.py 
54
55
56
57
58
59
60
61
62
63
64
65
def __init__(
    self,
    path: str,
    collection_name: str,
) -> None:
    self._path = path
    self._collection_name = collection_name
    self._client = chromadb.PersistentClient(path=path)
    self._collection = self._client.get_or_create_collection(
        name=collection_name,
        metadata={"hnsw:space": "cosine"},
    )

collection_info async 

collection_info() -> dict[str, Any]

Return collection metadata.

Returns:

Type Description
dict[str, Any]

Dict with name, path, and count keys.
Source code in symfonic/knowledge/adapters/chroma.py 
111
112
113
114
115
116
117
118
119
120
121
122
async def collection_info(self) -> dict[str, Any]:
    """Return collection metadata.

    Returns:
        Dict with ``name``, ``path``, and ``count`` keys.
    """
    count: int = await asyncio.to_thread(self._collection.count)
    return {
        "name": self._collection_name,
        "path": self._path,
        "count": count,
    }

search async 

search(query: str, limit: int = 3) -> list[KnowledgeFragment]

Query the collection using ChromaDB's built-in embedding.

Parameters:

Name Type Description Default
query str

Natural language search query.

 required
limit int

Maximum number of fragments to return.

 3

Returns:

Type Description
list[KnowledgeFragment]

List of KnowledgeFragment sorted by descending score.
Source code in symfonic/knowledge/adapters/chroma.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
async def search(
    self,
    query: str,
    limit: int = 3,
) -> list[KnowledgeFragment]:
    """Query the collection using ChromaDB's built-in embedding.

    Args:
        query: Natural language search query.
        limit: Maximum number of fragments to return.

    Returns:
        List of KnowledgeFragment sorted by descending score.
    """
    results: dict[str, Any] = await asyncio.to_thread(
        self._collection.query,
        query_texts=[query],
        n_results=limit,
    )

    fragments: list[KnowledgeFragment] = []
    if results and results.get("ids"):
        ids: list[str] = results["ids"][0]
        distances: list[float] = results.get("distances", [[]])[0]
        documents: list[str] = results.get("documents", [[]])[0]
        metadatas: list[dict[str, Any]] = results.get("metadatas", [[]])[0]

        for i, doc_id in enumerate(ids):
            score = 1.0 - distances[i] if i < len(distances) else 0.0
            doc = documents[i] if i < len(documents) else ""
            meta: dict[str, Any] = metadatas[i] if i < len(metadatas) else {}
            source: str = meta.get("source", doc_id)

            fragments.append(
                KnowledgeFragment(
                    content=doc,
                    source=source,
                    score=round(score, 4),
                    metadata=meta,
                )
            )

    return sorted(fragments, key=lambda f: f.score, reverse=True)