symfonic.knowledge.protocols

protocols

Protocol interfaces for the knowledge retrieval layer.

KnowledgeProvider is the single abstract interface that all RAG adapters must satisfy. Implementations can connect to ChromaDB, Pinecone, Qdrant, or any other vector store.

Architecture note: the protocol is @runtime_checkable so that adapters can be validated with isinstance() without requiring explicit inheritance.

Layering note (TA2.4): this module is the declared port of the knowledge capability -- T4.2.1's integration-adapter catalogue names symfonic.knowledge.protocols.KnowledgeProvider as the port symfonic.knowledge.adapters.chroma implements. The dependency matrix's integration row licenses "exactly the protocol module it implements and the contract types those protocols reference", and archcheck recognises a port only by module basename, so KnowledgeFragment is re-exported here rather than reached for in .models. There is one definition; this is the port surface naming it.

KnowledgeFragment dataclass

KnowledgeFragment(content: str, source: str, score: float, metadata: dict[str, Any] = dict())

A single piece of retrieved knowledge with source attribution.

Returned by every KnowledgeProvider.search() call. The score is always a cosine-similarity-derived value in [0.0, 1.0] where higher means more relevant.

content instance-attribute

content: str

The raw text of the retrieved document chunk.

metadata class-attribute instance-attribute

metadata: dict[str, Any] = field(default_factory=dict)

Arbitrary key-value metadata from the underlying store.

score instance-attribute

score: float

Similarity score in [0.0, 1.0]; higher is more relevant.

source instance-attribute

source: str

Human-readable source reference, e.g. 'LFT Art. 47' or 'RFC 7231 §5.3'.

format_citation

format_citation() -> str

Return a formatted citation string for prompt injection.

Example::

SOURCE [RFC 7231 §5.3]: Accept header field specifies acceptable...

Source code in src/symfonic/knowledge/models.py

def format_citation(self) -> str:
    """Return a formatted citation string for prompt injection.

    Example::

        SOURCE [RFC 7231 §5.3]: Accept header field specifies acceptable...
    """
    return f"SOURCE [{self.source}]: {self.content}"

KnowledgeProvider

Bases: Protocol

Abstract interface for semantic knowledge retrieval (RAG).

Concrete implementations connect to vector databases (ChromaDB, Pinecone, Qdrant, etc.) and return scored knowledge fragments.

All methods are async. Synchronous backends must wrap blocking calls with asyncio.to_thread.

collection_info async

collection_info() -> dict[str, Any]

Return metadata about the connected collection.

Returns:

Type	Description
dict[str, Any]	Dict with at least name and count keys.

Source code in src/symfonic/knowledge/protocols.py

async def collection_info(self) -> dict[str, Any]:
    """Return metadata about the connected collection.

    Returns:
        Dict with at least ``name`` and ``count`` keys.
    """
    ...

search async

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

Search for knowledge fragments matching the query.

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 src/symfonic/knowledge/protocols.py

async def search(
    self,
    query: str,
    limit: int = 3,
) -> list[KnowledgeFragment]:
    """Search for knowledge fragments matching the query.

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

    Returns:
        List of KnowledgeFragment sorted by descending score.
    """
    ...