Initialize ingestion pipeline. Args: config: Ingestion configuration documents_folder: Folder containing mar

Initialize cache.

Run the crawl tool. Args: url: URL to crawl extract_structured: Whether to extract structured data

Return (base64_data, mime_type) for a local file.

Authorization: Bearer header takes priority; falls back to access_token cookie.

Construct a full key from namespace hierarchy + key.

Check if data matches the filter criteria.

Synchronous fallback.

Batch multimodal calls. Each item in `items` should have keys: text, and one of image_path | image_url | image_b64.

Index multiple documents concurrently with a bounded in-flight window. Raises exceptions immediately if any task fails.

Batch async text calls. Uses LangChain's native abatch which honours max_concurrency.

Call PageIndex chat completion API in non-streaming mode.

Create an explicit Gemini context cache for large reusable prompts. Use this for shared context that would otherwise be re-sent on every req

Example task: Add two numbers.

Add item(s) to a Bloom filter (BF.ADD or BF.MADD). Args: redis: Redis client instance filter_name: Name of the filter val

Embed multiple strings, respecting concurrency limits. GoogleGenerativeAIEmbeddings.embed_documents handles batching internally. Args:

Embed a single string asynchronously. Args: text: The text to embed. model: Embedding model. Defaults to Gemini embeddings.

Return a configured chat model, defaulting all behavior from settings.

Adjust guard-managed headers for streaming responses.

Stream chat completion chunks from PageIndex as an async generator. Bridges the SDK's sync iterator to async code using asyncio.to_thread and a q

Async generator that yields text tokens as they arrive. Usage in FastAPI:: async def sse_endpoint(): async for chunk in

Return a model augmented with structured output. Invoke like a normal chain; output is a validated Pydantic model. Example:: cl

Assemble the full message list for an LLM call. Steps: 1. Retrieve relevant Graphiti context (if include_long_term_memory) 2. Filter

Build the contract ingestion graph once during application startup.

Lightweight stack for development / testing.

Build graph config with lifespan-owned clients attached.

Build the reconciliation graph once during application startup.

Build and compile the Agent Saul LangGraph.

Build all tools once. Call at lifespan startup only.

Return tools matching ALL given tags.

Get user with automatic cache fallback pattern. This pattern: 1. Tries cache 2. If miss, tries DB 3. If DB fails too, returns None