Knowledge Base Service - Examples

This page provides practical code examples for implementing knowledge base operations. For broader documentation and concepts, see the Knowledge Base Documentation.

Content Upload

Upload from Memory (Recommended)

#kb_service_upload_bytes

content_bytes = b"Your file content here"
content = kb_service.upload_content_from_bytes(
    content=content_bytes,
    content_name="document.txt",
    mime_type="text/plain",
    scope_id=scope_id,
    metadata={"category": "documentation", "version": "1.0"}
)

Upload from File

#kb_service_upload_from_file

# Configure ingestion settings
content = kb_service.upload_content(
    path_to_content=str(file_path),
    content_name=Path(file_path).name,
    mime_type="text/plain",
    scope_id=scope_id,
    skip_ingestion=False,  # Process the content for search
    metadata={"department": "legal", "classification": "confidential"}
)

Make Uploaded Document Available to User

#kb-service-make-document-available

uploaded_content = kb_service.upload_content(
        path_to_content=str(output_filepath),
        content_name=output_filepath.name,
        mime_type=str(mimetypes.guess_type(output_filepath)[0]),
        chat_id=payload.chat_id,
        skip_ingestion=skip_ingestion,  # Usually True for generated files
    )

reference = ContentReference(
    id=content.id,
    sequence_number=1,
    message_id=message_id,
    name=filename,
    source=payload.name,
    source_id=chat_id,
    url=f"unique://content/{uploaded_content.id}",  # Special URL format for content
)

self.chat_service.modify_assistant_message(
                content="Please find the translated document below in the references.",
                references=[reference],
                set_completed_at=True,
            )

Full Examples Upload (Click to expand)

Upload from MemoryUpload from File

# %%
from pathlib import Path

from dotenv import dotenv_values

from unique_toolkit import (
    KnowledgeBaseService,
)

kb_service = KnowledgeBaseService.from_settings()
demo_env_vars = dotenv_values(Path(__file__).parent / "demo.env")
scope_id = demo_env_vars.get("UNIQUE_SCOPE_ID") or "unknown"
content_bytes = b"Your file content here"
content = kb_service.upload_content_from_bytes(
    content=content_bytes,
    content_name="document.txt",
    mime_type="text/plain",
    scope_id=scope_id,
    metadata={"category": "documentation", "version": "1.0"},
)

# %%
from pathlib import Path

from dotenv import dotenv_values

from unique_toolkit import (
    KnowledgeBaseService,
)

kb_service = KnowledgeBaseService.from_settings()
demo_env_vars = dotenv_values(Path(__file__).parent / "demo.env")
scope_id = demo_env_vars.get("UNIQUE_SCOPE_ID") or "unknown"
file_path = Path(__file__).parent / "test.txt"
# Configure ingestion settings
content = kb_service.upload_content(
    path_to_content=str(file_path),
    content_name=Path(file_path).name,
    mime_type="text/plain",
    scope_id=scope_id,
    skip_ingestion=False,  # Process the content for search
    metadata={"department": "legal", "classification": "confidential"},
)

Mirror a local folder (structure + files, no ingestion)

You can recreate a directory tree in the knowledge base by creating the corresponding folder paths with :class:~unique_toolkit.experimental.resources.content_folder.service.ContentFolder, then uploading each file into the leaf scope for its directory with skip_ingestion=True. Files are stored but not sent through the ingestion pipeline (no chunking / vector index for search).

Scope and paths

kb_root_path must be an absolute knowledge-base path (leading /). Adjust LOCAL_ROOT to any folder on your machine. The sample uses ./sample_tree relative to the current working directory when you run the script.

#kb-mirror-local-imports

from __future__ import annotations

import mimetypes
from pathlib import Path

from unique_toolkit.experimental.resources.content_folder import ContentFolder
from unique_toolkit.services.knowledge_base import KnowledgeBaseService

#kb-mirror-local-mirror-fn

def mirror_local_folder_to_kb(
    *,
    local_root: Path,
    kb_root_path: str,
    kb_service: KnowledgeBaseService,
    folder_service: ContentFolder,
) -> None:
    """Upload every file under *local_root*, preserving subfolders, without ingestion."""
    local_root = local_root.resolve()
    if not local_root.is_dir():
        msg = f"Not a directory: {local_root}"
        raise NotADirectoryError(msg)

    kb_prefix = kb_root_path if kb_root_path.startswith("/") else f"/{kb_root_path}"
    kb_prefix = kb_prefix.rstrip("/") or "/"

    scope_by_kb_dir: dict[str, str] = {}

    def scope_for_kb_dir(kb_dir: str) -> str:
        if kb_dir not in scope_by_kb_dir:
            created = folder_service.create(paths=kb_dir)
            scope_by_kb_dir[kb_dir] = created[-1].id
        return scope_by_kb_dir[kb_dir]

    _ = scope_for_kb_dir(kb_prefix)

    for file_path in sorted(p for p in local_root.rglob("*") if p.is_file()):
        rel = file_path.relative_to(local_root)
        parts = rel.parts
        parent_parts = parts[:-1]
        filename = parts[-1]
        kb_dir = kb_prefix + ("/" + "/".join(parent_parts) if parent_parts else "")
        scope_id = scope_for_kb_dir(kb_dir)
        mime = mimetypes.guess_type(filename)[0] or "application/octet-stream"
        kb_service.upload_content(
            path_to_content=str(file_path),
            content_name=filename,
            mime_type=mime,
            scope_id=scope_id,
            skip_ingestion=True,
        )

#kb-mirror-local-run

if __name__ == "__main__":
    LOCAL_ROOT = Path("./sample_tree").resolve()
    KB_ROOT = "/LocalMirror/MyProject"

    mirror_local_folder_to_kb(
        local_root=LOCAL_ROOT,
        kb_root_path=KB_ROOT,
        kb_service=KnowledgeBaseService.from_settings(),
        folder_service=ContentFolder.from_settings(),
    )

#kb-mirror-main / file: docs/.python_files/kb_mirror_local_folder_skip_ingest.py

<<kb-mirror-local-imports>>

<<kb-mirror-local-mirror-fn>>

<<kb-mirror-local-run>>

Mirror local folder — full script

Mirror local folder (skip ingestion)

# %%
from __future__ import annotations

import mimetypes
from pathlib import Path

from unique_toolkit.experimental.resources.content_folder import ContentFolder
from unique_toolkit.services.knowledge_base import KnowledgeBaseService


def mirror_local_folder_to_kb(
    *,
    local_root: Path,
    kb_root_path: str,
    kb_service: KnowledgeBaseService,
    folder_service: ContentFolder,
) -> None:
    """Upload every file under *local_root*, preserving subfolders, without ingestion."""
    local_root = local_root.resolve()
    if not local_root.is_dir():
        msg = f"Not a directory: {local_root}"
        raise NotADirectoryError(msg)

    kb_prefix = kb_root_path if kb_root_path.startswith("/") else f"/{kb_root_path}"
    kb_prefix = kb_prefix.rstrip("/") or "/"

    scope_by_kb_dir: dict[str, str] = {}

    def scope_for_kb_dir(kb_dir: str) -> str:
        if kb_dir not in scope_by_kb_dir:
            created = folder_service.create(paths=kb_dir)
            scope_by_kb_dir[kb_dir] = created[-1].id
        return scope_by_kb_dir[kb_dir]

    _ = scope_for_kb_dir(kb_prefix)

    for file_path in sorted(p for p in local_root.rglob("*") if p.is_file()):
        rel = file_path.relative_to(local_root)
        parts = rel.parts
        parent_parts = parts[:-1]
        filename = parts[-1]
        kb_dir = kb_prefix + ("/" + "/".join(parent_parts) if parent_parts else "")
        scope_id = scope_for_kb_dir(kb_dir)
        mime = mimetypes.guess_type(filename)[0] or "application/octet-stream"
        kb_service.upload_content(
            path_to_content=str(file_path),
            content_name=filename,
            mime_type=mime,
            scope_id=scope_id,
            skip_ingestion=True,
        )


if __name__ == "__main__":
    LOCAL_ROOT = Path("./sample_tree").resolve()
    KB_ROOT = "/LocalMirror/MyProject"

    mirror_local_folder_to_kb(
        local_root=LOCAL_ROOT,
        kb_root_path=KB_ROOT,
        kb_service=KnowledgeBaseService.from_settings(),
        folder_service=ContentFolder.from_settings(),
    )

Verify the result

To inspect the resulting folder layout after a mirror, use :class:~unique_toolkit.experimental.components.content_tree.service.ContentTree from the experimental content_tree subpackage — see the content tree example. A combined mirror + verify example will follow in a small follow-up PR.

Content Download

Download to Memory (Recommended)

#kb_service_download_bytes

# Download content as bytes
content_bytes = kb_service.download_content_to_bytes(
    content_id=content_id or "unknown",
)

# Process in memory
text = ""
with io.BytesIO(content_bytes) as file_like:
    text = file_like.read().decode("utf-8")

print(text)

Download to Temporary File

#kb_service_download_file

# Download to secure temporary file

filename = "my_testfile.txt"
temp_file_path = kb_service.download_content_to_file(
    content_id=content_id,
    output_filename=filename,
    output_dir_path=Path(tempfile.mkdtemp())  # Use secure temp directory
)

try:
    # Process the file
    with open(temp_file_path, 'rb') as file:
        text = file.read().decode("utf-8")
        print(text) 
finally:
    # Always clean up temporary files
    if temp_file_path.exists():
        temp_file_path.unlink()
    # Clean up the temporary directory
    temp_file_path.parent.rmdir()

Full Examples Download (Click to expand)

Download to MemoryDownload to File

# %%
import io
from pathlib import Path

from dotenv import dotenv_values

from unique_toolkit import (
    KnowledgeBaseService,
)

kb_service = KnowledgeBaseService.from_settings()
demo_env_vars = dotenv_values(Path(__file__).parent / "demo.env")
content_id = demo_env_vars.get("UNIQUE_CONTENT_ID") or "unknown"
# Download content as bytes
content_bytes = kb_service.download_content_to_bytes(
    content_id=content_id or "unknown",
)

# Process in memory
text = ""
with io.BytesIO(content_bytes) as file_like:
    text = file_like.read().decode("utf-8")

print(text)

# %%
import tempfile
from pathlib import Path

from dotenv import dotenv_values

from unique_toolkit import (
    KnowledgeBaseService,
)

kb_service = KnowledgeBaseService.from_settings()
demo_env_vars = dotenv_values(Path(__file__).parent / "demo.env")
content_id = demo_env_vars.get("UNIQUE_CONTENT_ID") or "unknown"
# Download to secure temporary file

filename = "my_testfile.txt"
temp_file_path = kb_service.download_content_to_file(
    content_id=content_id,
    output_filename=filename,
    output_dir_path=Path(tempfile.mkdtemp()),  # Use secure temp directory
)

try:
    # Process the file
    with open(temp_file_path, "rb") as file:
        text = file.read().decode("utf-8")
        print(text)
finally:
    # Always clean up temporary files
    if temp_file_path.exists():
        temp_file_path.unlink()
    # Clean up the temporary directory
    temp_file_path.parent.rmdir()

Content Deletion

#kb_service_delete_content

kb_service.delete_content(
    content_id=content.id
)

Full Examples Content Deletion (Click to expand)

Content Deletion

# %%
from pathlib import Path

from dotenv import dotenv_values

from unique_toolkit import (
    KnowledgeBaseService,
)

kb_service = KnowledgeBaseService.from_settings()
demo_env_vars = dotenv_values(Path(__file__).parent / "demo.env")
scope_id = demo_env_vars.get("UNIQUE_SCOPE_ID") or "unknown"
content_bytes = b"Your file content here"
content = kb_service.upload_content_from_bytes(
    content=content_bytes,
    content_name="document.txt",
    mime_type="text/plain",
    scope_id=scope_id,
    metadata={"category": "documentation", "version": "1.0"},
)
kb_service.delete_content(content_id=content.id)

Content Search

Semantic Search (Vector-Based)

#kb_service_vector_search

# Search for content using vector similarity
content_chunks = kb_service.search_content_chunks(
    search_string="Harry Potter",
    search_type=ContentSearchType.VECTOR,
    limit=10,
    score_threshold=0.7,  # Only return results with high similarity
    scope_ids=[scope_id]
)

print(f"Found {len(content_chunks)} relevant chunks")
for i, chunk in enumerate(content_chunks[:3]):
    print(f"  {i+1}. {chunk.text[:100]}...")

Combined Search (Hybrid)

#kb_service_combined_search

# Combined semantic and keyword search for best results
content_chunks = kb_service.search_content_chunks(
    search_string="Harry Potter",
    search_type=ContentSearchType.COMBINED,
    limit=15,
    search_language="english",
    scope_ids=[scope_id],  # Limit to specific scopes if configured
)

print(f"Combined search found {len(content_chunks)} chunks")

Content File Search

#kb_service_content_search

# Search for specific content files
contents = kb_service.search_contents(
    where={"title": {"contains": "manual"}},
)

Full Examples

Full Examples Content Search (Click to expand)

Vector SearchCombined SearchContent Search

# %%
from pathlib import Path

from dotenv import dotenv_values

from unique_toolkit import (
    KnowledgeBaseService,
)
from unique_toolkit.content.schemas import (
    ContentSearchType,
)

kb_service = KnowledgeBaseService.from_settings()
demo_env_vars = dotenv_values(Path(__file__).parent / "demo.env")
scope_id = demo_env_vars.get("UNIQUE_SCOPE_ID") or "unknown"
# Search for content using vector similarity
content_chunks = kb_service.search_content_chunks(
    search_string="Harry Potter",
    search_type=ContentSearchType.VECTOR,
    limit=10,
    score_threshold=0.7,  # Only return results with high similarity
    scope_ids=[scope_id],
)

print(f"Found {len(content_chunks)} relevant chunks")
for i, chunk in enumerate(content_chunks[:3]):
    print(f"  {i + 1}. {chunk.text[:100]}...")

# %%
from pathlib import Path

from dotenv import dotenv_values

from unique_toolkit import (
    KnowledgeBaseService,
)
from unique_toolkit.content.schemas import (
    ContentSearchType,
)

kb_service = KnowledgeBaseService.from_settings()
demo_env_vars = dotenv_values(Path(__file__).parent / "demo.env")
scope_id = demo_env_vars.get("UNIQUE_SCOPE_ID") or "unknown"
# Combined semantic and keyword search for best results
content_chunks = kb_service.search_content_chunks(
    search_string="Harry Potter",
    search_type=ContentSearchType.COMBINED,
    limit=15,
    search_language="english",
    scope_ids=[scope_id],  # Limit to specific scopes if configured
)

print(f"Combined search found {len(content_chunks)} chunks")

# %%
from pathlib import Path

from dotenv import dotenv_values

from unique_toolkit import (
    KnowledgeBaseService,
)

kb_service = KnowledgeBaseService.from_settings()
demo_env_vars = dotenv_values(Path(__file__).parent / "demo.env")
scope_id = demo_env_vars.get("UNIQUE_SCOPE_ID") or "unknown"
# Search for specific content files
contents = kb_service.search_contents(
    where={"title": {"contains": "manual"}},
)