> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agno.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Error Handling: Production Patterns

> Patterns for robust knowledge ingestion.

```python error_handling.py theme={null}
"""
Error Handling: Production Patterns
=====================================
Production knowledge systems need to handle failures gracefully:
- Content that fails to load
- Vector DB connection issues
- Large document batches with partial failures

This example shows patterns for robust knowledge ingestion.
"""

import asyncio
import logging

from agno.agent import Agent
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.knowledge.knowledge import Knowledge
from agno.models.openai import OpenAIResponses
from agno.vectordb.qdrant import Qdrant
from agno.vectordb.search import SearchType

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Setup
# ---------------------------------------------------------------------------

qdrant_url = "http://localhost:6333"

knowledge = Knowledge(
    vector_db=Qdrant(
        collection="error_handling_demo",
        url=qdrant_url,
        search_type=SearchType.hybrid,
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    knowledge=knowledge,
    search_knowledge=True,
    markdown=True,
)

# ---------------------------------------------------------------------------
# Run Demo
# ---------------------------------------------------------------------------

if __name__ == "__main__":

    async def main():
        # --- 1. Safe insert with skip_if_exists ---
        print("\n" + "=" * 60)
        print("PATTERN 1: Idempotent inserts with skip_if_exists")
        print("=" * 60 + "\n")

        # Safe to call multiple times - won't re-process existing content
        await knowledge.ainsert(
            name="Recipes",
            url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
            skip_if_exists=True,
        )
        print("Insert completed (skipped if already exists)")

        # --- 2. Batch with mixed valid/invalid sources ---
        print("\n" + "=" * 60)
        print("PATTERN 2: Batch insert with error logging")
        print("=" * 60 + "\n")

        sources = [
            {"name": "Valid", "text_content": "This will succeed."},
            {"name": "Also Valid", "text_content": "This will also succeed."},
        ]

        for source in sources:
            try:
                await knowledge.ainsert(**source)
                print("Inserted: %s" % source["name"])
            except Exception as e:
                logger.error("Failed to insert %s: %s", source["name"], e)

        # --- 3. Verify knowledge is usable ---
        print("\n" + "=" * 60)
        print("PATTERN 3: Verify knowledge after ingestion")
        print("=" * 60 + "\n")

        agent.print_response("What do you know?", stream=True)

    asyncio.run(main())
```

## Run the Example

<Steps>
  <Snippet file="create-venv-step.mdx" />

  <Step title="Install dependencies">
    ```bash theme={null}
    uv pip install -U agno fastembed openai qdrant-client
    ```
  </Step>

  <Step title="Export your OpenAI API key">
    <CodeGroup>
      ```bash Mac/Linux theme={null}
      export OPENAI_API_KEY="your_openai_api_key_here"
      ```

      ```bash Windows theme={null}
      $Env:OPENAI_API_KEY="your_openai_api_key_here"
      ```
    </CodeGroup>
  </Step>

  <Step title="Run Qdrant">
    ```bash theme={null}
    docker run -d --name qdrant -p 6333:6333 qdrant/qdrant:latest
    ```
  </Step>

  <Step title="Run the example">
    Save the code above as `error_handling.py`, then run:

    ```bash theme={null}
    python error_handling.py
    ```
  </Step>
</Steps>

Full source: [cookbook/07\_knowledge/03\_production/04\_error\_handling.py](https://github.com/agno-agi/agno/blob/main/cookbook/07_knowledge/03_production/04_error_handling.py)