Chatbot Framework: Complete Production-Ready AI Chatbot System

A sophisticated, production-grade Python framework for building embeddable AI chatbots with optional capabilities for SQL databases, web access, RAG (Retrieval-Augmented Generation), validation, reflection, and memory management.

Current Status: 232 tests passing | Type-safe (Pyright strict mode) | Production-ready

---

Table of Contents

1. What is It?
2. Key Features
3. Installation
4. Quick Start
5. Architecture Overview
6. Core Concepts
7. Capabilities
8. How to Use
9. Examples
10. API Reference
11. Configuration & Security
12. Observability
13. Advanced Usage
14. Integration with Existing Applications
15. Best Practices
16. Troubleshooting

---

What is It?

The Chatbot Framework is an embeddable, modular Python framework for creating sophisticated AI chatbots. Unlike monolithic chatbot solutions, this framework is designed to be:

• Modular: Use only the capabilities you need
• Type-Safe: Full type annotations, Pyright strict mode compatible
• Production-Ready: Security hardening, observability, configuration management
• Flexible: Works with 0 to all capabilities enabled
• Observable: Structured event tracking for debugging and monitoring
• Extensible: Protocol-based interfaces for adding custom capabilities

What Can It Do?

• Agentic Loop: Plan → Execute → Observe → Replan cycle
• Multi-Capability Orchestration: Automatically route queries to SQL, Web, or RAG
• Quality Control: Validate outputs before returning to users
• Reflection: Self-evaluate and improve answers
• Memory: Maintain conversation history and long-term knowledge
• Security: Credential filtering, authorization, centralized limits
• Observability: Structured events for all operations

---

Key Features

✅ Agentic Architecture (Phases 1-5)

• Plan generation using LLM
• Step-by-step execution tracking
• Self-healing replanning (max 3 attempts)
• Safety limits: max 10 steps per plan, 60s timeout
• Comprehensive execution state tracking

✅ RAG (Retrieval-Augmented Generation) - Phase 6

• Document loading and chunking
• Vector embeddings via Qdrant
• Semantic search over documents
• Grounding answers in source material

✅ SQL Database Access - Phase 7

• SELECT-only queries (write operations blocked)
• Validated query parameters
• Schema introspection
• Read-only connection enforcement

✅ Web Capabilities - Phase 8

• HTTP fetch with timeout protection
• Mock and real HTTP clients
• Multi-capability orchestration
• Automatic routing based on query keywords

✅ Quality Control - Phase 9

• 5 built-in validators:
  • Relevance: Keyword overlap checking
  • Grounding: Source material validation
  • Tool Input: Parameter validation
  • Output Format: Length and format checks
  • Composite: Priority-based aggregation
• Reflection: Self-evaluation and refinement (max 3 attempts)
• Hard gate: Prevents unsafe outputs from reaching users

✅ Memory Management - Phase 9

• Conversation history tracking
• Multi-turn context preservation
• Application state management
• Long-term knowledge storage
• Pattern learning

✅ Production Packaging - Phase 10

• Zero heavy dependencies (base)
• Optional extras for each capability
• Structured observability with pluggable sinks
• Centralized configuration and security
• 5 production example applications

---

Installation

Base Installation (Pydantic only, no heavy dependencies)

pip install chatbot-framework

With Optional Capabilities

# OpenAI support
pip install chatbot-framework[openai]

# Vector database (Qdrant) + embeddings
pip install chatbot-framework[qdrant]

# SQL database support
pip install chatbot-framework[sql]

# Web client (aiohttp)
pip install chatbot-framework[web]

# Document processing (PDF, DOCX)
pip install chatbot-framework[documents]

# All capabilities
pip install chatbot-framework[all]

Development Installation

# Install with all dependencies and development tools
pip install -e ".[dev]"

# Run tests
pytest tests/ -v

# Type checking
pyright src/

# Linting
ruff check src/

Verify Installation

# Test base package
from chatbot_framework import Chatbot, Capability
print("✓ Base package installed")

# Test optional packages
from chatbot_framework.config import FrameworkConfig
from chatbot_framework.observability import EventEmitter
from chatbot_framework.memory import MemoryManager
print("✓ All modules available")

---

Quick Start

Minimal Chatbot (Text-Only)

import asyncio
from chatbot_framework import Chatbot

async def main():
    # Create chatbot with no external capabilities
    chatbot = Chatbot(capabilities=[])
    
    # Chat with the bot
    response = await chatbot.chat("Hello, how are you?")
    print(response.text)

asyncio.run(main())

Chatbot with RAG (Knowledge Base)

import asyncio
from chatbot_framework import Chatbot
from chatbot_framework.rag import RAGCapability, VectorStore

async def main():
    # Setup RAG
    vector_store = VectorStore()  # Mocked for this example
    rag = RAGCapability(vector_store=vector_store)
    
    # Create chatbot with RAG
    chatbot = Chatbot(capabilities=[rag])
    
    # Query with grounding in documents
    response = await chatbot.chat("What's in our documentation?")
    print(response.text)

asyncio.run(main())

Enterprise Chatbot with SQL + Web

import asyncio
from chatbot_framework import Chatbot
from chatbot_framework.sql import SQLCapability, SQLConnection
from chatbot_framework.web import WebCapability, SimpleWebClient

async def main():
    # Setup capabilities
    db = SQLConnection("sqlite:///company.db", read_only=True)
    sql = SQLCapability(connection=db)
    web = WebCapability(client=SimpleWebClient())
    
    # Create enterprise chatbot
    chatbot = Chatbot(capabilities=[sql, web])
    
    # Multi-capability queries work automatically
    response = await chatbot.chat(
        "Compare our sales with industry benchmarks from the web"
    )
    print(response.text)

asyncio.run(main())

Full-Featured Agent with All Capabilities

import asyncio
from chatbot_framework import (
    Chatbot,
    MemoryManager,
    CompositeValidator,
    ReflectionCapability,
    configure_production,
)

async def main():
    # Configure framework for production
    configure_production()
    
    # Setup all capabilities
    capabilities = [sql, web, rag, reflection]
    
    # Create advanced agent
    agent = Chatbot(capabilities=capabilities)
    
    # Multi-turn conversation with memory
    response1 = await agent.chat("What's our revenue?")
    response2 = await agent.chat("Compare with competitors")
    print(response1.text)
    print(response2.text)

asyncio.run(main())

---

Architecture Overview

High-Level Flow (Top-Level Orchestration)

User Query
    ↓
[Chatbot/Orchestrator]
    ↓
[Planner] — Top-Level Decision Maker
    ├─ Analyzes query
    ├─ Determines required capabilities
    ├─ Creates multi-step execution plan
    ├─ Defines step dependencies
    └─ Returns structured Plan
    ↓
[Executor] — Step-by-Step Execution
    ├─ Execute independent steps (parallel where possible)
    ├─ Execute dependent steps (respecting order)
    ├─ Route each step to appropriate capability:
    │   ├─ [SQL Capability] → Query database
    │   ├─ [Web Capability] → Fetch web data
    │   ├─ [RAG Capability] → Search documents
    │   ├─ [Tool Registry] → Call custom tools
    │   ├─ [Synthesizer] → Combine multi-source results
    │   └─ [LLM] → Generate answers
    └─ Collect results
    ↓
[Validators] → Check output quality (hard gate)
    ├─ Relevance Validator
    ├─ Grounding Validator
    ├─ Tool Input Validator
    ├─ Output Format Validator
    └─ Composite Validator
    ↓
[Reflection] → Self-evaluate and improve
    ↓
[Memory] → Store conversation context
    ↓
Response to User

Multi-Step Planning Examples

The Planner creates different plan structures based on query complexity:

Example 1: Simple Query (No External Data)

Query: "Explain dependency injection."
Plan:
  - Step 1: [LLM] Generate answer directly
  - Final: Return LLM response

Example 2: Single-Capability Query

Query: "What does our employee handbook say about leave?"
Plan:
  - Step 1: [RAG] Search internal documents → Result
  - Step 2: [LLM] Generate answer (depends on Step 1)
  - Final: Return answer

Example 3: Multi-Capability Query (Independent Steps)

Query: "Compare our Q2 sales with industry benchmarks."
Plan:
  - Step 1: [SQL] Query company Q2 sales → Result
  - Step 2: [Web] Fetch industry benchmarks → Result
  - Step 3: [Synthesizer] Combine SQL + Web results → Synthesis
  - Step 4: [LLM] Generate final answer (depends on Step 3)
  - Final: Return answer

Example 4: Multi-Capability Query with Dependencies

Query: "Find our product with largest sales decline and explain why using market data."
Plan:
  - Step 1: [SQL] Find product with largest decline → Product ID
  - Step 2: [Web] Search market for that product (depends on Step 1) → Market Data
  - Step 3: [RAG] Search internal docs for that product (parallel with Step 2) → Internal Data
  - Step 4: [Synthesizer] Combine SQL + Web + RAG → Synthesis
  - Step 5: [LLM] Generate final answer (depends on Step 4)
  - Final: Return answer

Component Layers

┌────────────────────────────────────────────────────┐
│   User/Application Layer                           │
├────────────────────────────────────────────────────┤
│   Chatbot (Main Interface)                         │
├────────────────────────────────────────────────────┤
│   Orchestration                                    │
│   └─ Planner (Top-Level Decision Maker)            │
│      ├─ Query Analysis                             │
│      ├─ Capability Selection & Routing             │
│      ├─ Dependency Management                      │
│      └─ Returns structured Plan                    │
├────────────────────────────────────────────────────┤
│   Execution                                        │
│   ├─ Executor (Step-by-step execution)             │
│   ├─ Synthesizer (Multi-source synthesis)          │
│   └─ Tool Registry                                 │
├────────────────────────────────────────────────────┤
│   Capabilities (Optional, Composable)              │
│   ├─ SQL Capability (Query database)               │
│   ├─ Web Capability (Fetch web data)               │
│   ├─ RAG Capability (Search documents)             │
│   └─ Custom Capabilities                           │
├────────────────────────────────────────────────────┤
│   Quality Control                                  │
│   ├─ Validators (Hard gates)                       │
│   └─ Reflection (Self-refinement)                  │
├────────────────────────────────────────────────────┤
│   Support Systems                                  │
│   ├─ Memory (Context preservation)                 │
│   ├─ Configuration (Limits/Security)               │
│   └─ Observability (Events/Logging)                │
└────────────────────────────────────────────────────┘

---

Core Concepts

1. Chatbot (Main Interface)

The Chatbot class is your primary interface to the framework.

from chatbot_framework import Chatbot

chatbot = Chatbot(
    capabilities=[...],           # Optional capabilities
    max_steps=10,                 # Max planning steps
    max_replans=3,                # Max replan attempts
    max_execution_time=60,        # Timeout in seconds
)

# Chat (main method)
response = await chatbot.chat("Your question")
print(response.text)              # Generated response
print(response.reasoning)         # Internal reasoning
print(response.sources)           # Used sources

2. Planner (Top-Level Decision Maker)

The Planner is the core orchestration component that determines which capabilities are needed and how to combine them.

The Planner:

• Analyzes the user query to understand intent
• Routes each request through one or more capabilities (SQL, RAG, Web, Tools)
• Creates a structured execution plan with dependencies
• Synthesizes results from multiple sources (when needed)

from chatbot_framework.planning import Planner
from chatbot_framework.models import Plan, Step

# Planner receives LLM for decision making
planner = Planner(llm)

# Planner creates different plans based on query type:

# Plan 1: Simple query (no external data needed)
# Plan: [LLM] → Answer

# Plan 2: Single-capability query
# Plan: [SQL] → [Synthesize with LLM] → Answer

# Plan 3: Multi-capability query (independent steps)
# Plan: [SQL, Web, RAG] (parallel) → [Synthesize] → Answer

# Plan 4: Multi-capability with dependencies
# Plan: [SQL find product] → [Web search that product] → [Synthesize] → Answer

Key Features:

• LLM-powered decision making (not keyword-based routing)
• Automatic capability selection and routing
• Dependency graph management
• Synthesis strategy selection (combine, compare, explain)
• Safe limits on planning steps

3. Synthesizer (Multi-Source Combination)

The Synthesizer combines results from multiple capabilities into a cohesive answer.

from chatbot_framework.orchestration import Synthesizer

synthesizer = Synthesizer(llm)

# Synthesize multi-source results
result = await synthesizer.synthesize(
    original_query="Compare our sales with industry benchmarks",
    capability_results={
        "sql_query": "Our Q2 sales: $5M",
        "web_fetch": "Industry average Q2: $3.5M",
        "rag_query": "Our target: $6M by Q4",
    },
    strategy="compare",  # compare, explain, or combine
)

print(result.synthesized_answer)
print(result.sources_used)
print(result.confidence)

4. Executor (Step-by-Step Execution)

The Executor runs the plan step-by-step, respecting dependencies and executing independent steps in parallel where possible.

from chatbot_framework.runtime import Executor

executor = Executor(
    llm=llm,
    planner=planner,
    max_tool_calls=5,
    max_execution_time=30.0,
    max_replans=3,
)

# Execute plan with event streaming
async for event in executor.execute(plan, request):
    print(f"[{event.event_type}] {event.message}")
    if event.data:
        print(f"  Data: {event.data}")

Features:

• Dependency-aware execution
• Parallel execution where possible
• Replanning support (Phase 5+)
• Comprehensive event tracking
• Safety limits enforcement

5. Capabilities (Optional)

Capabilities are composable modules that extend chatbot functionality. Each is optional and independently usable.

# Works with 0, 1, 2, or all capabilities
chatbot = Chatbot(
    capabilities=[
        SQLCapability(...),       # Optional
        WebCapability(...),       # Optional
        RAGCapability(...),       # Optional
    ]
)

# Or minimal (no external access)
chatbot = Chatbot(capabilities=[])

Built-in Capabilities:

• SQLCapability: Query databases (SELECT only)
• WebCapability: Fetch web resources
• RAGCapability: Search documents
• ReflectionCapability: Self-evaluation

6. Validators (Quality Control)

Validators form a hard gate - they prevent unsafe outputs from reaching users.

from chatbot_framework.validation import CompositeValidator

validator = CompositeValidator([
    RelevanceValidator(),
    GroundingValidator(),
    ToolInputValidator(),
    OutputFormatValidator(),
])

result = validator.validate(data)
if not result.is_valid():
    return "Cannot process this safely"

Validation Types:

• Relevance: Keywords in output match user query
• Grounding: Answer references source material
• Tool Input: Parameters are valid
• Output Format: Length and format checks
• Composite: Aggregates all validations

7. Reflection (Self-Refinement)

Reflection allows the chatbot to evaluate its own output and improve it.

from chatbot_framework.reflection import ReflectionCapability

reflection = ReflectionCapability(max_attempts=3)

# Evaluate quality
quality = reflection.invoke("evaluate", {
    "output": "Some answer",
    "sources": ["doc1", "doc2"],
})

# Get improvement suggestions
suggestions = reflection.invoke("suggest_improvement", {
    "output": "Some answer",
    "feedback": "Not grounded enough",
})

# Determine if should retry
should_retry = reflection.invoke("retry", {
    "attempt": 1,
    "quality_score": 0.6,
})

8. Memory (Context Preservation)

Multi-tier memory system for maintaining context across turns.

from chatbot_framework.memory import MemoryManager

memory = MemoryManager()

# Conversation turns
memory.add_conversation_turn("user", "How are you?")
memory.add_conversation_turn("assistant", "I'm well, thanks!")
context = memory.get_conversation_context(num_turns=5)

# State tracking
memory.update_state(execution_count=1, error_count=0)
state = memory.get_state()

# Long-term facts
memory.store_fact("company_revenue", "$100M")
revenue = memory.recall_fact("company_revenue")

# Pattern learning
memory.learn_pattern("user_preference", "verbose_answers")

9. Execution Plan (Multi-Step Structure)

Plans have multiple steps with optional dependencies, not just routing to one capability.

from chatbot_framework.models import Plan, Step, StepStatus

# Capability steps (execute capabilities)
sql_step = Step(
    id="step_0",
    action="capability",
    description="Query company sales database",
    capability_name="sql_query",
    params={"query": "SELECT SUM(revenue) FROM sales WHERE quarter='Q2'"},
)

# Dependent step (uses result from sql_step)
web_step = Step(
    id="step_1",
    action="capability",
    description="Search market for benchmark data",
    capability_name="web_fetch",
    depends_on=["step_0"],  # Depends on SQL step result
    params={"query": "industry Q2 sales benchmark"},
)

# Synthesis step (combines SQL + Web results)
synthesis_step = Step(
    id="step_2",
    action="synthesize",
    description="Combine SQL and web data",
    capability_name="synthesizer",
    depends_on=["step_0", "step_1"],  # Depends on both previous steps
    params={"strategy": "compare"},
)

# Answer step (uses synthesis)
answer_step = Step(
    id="step_3",
    action="answer",
    description="Generate final answer",
    depends_on=["step_2"],  # Depends on synthesis
)

plan = Plan(steps=[sql_step, web_step, synthesis_step, answer_step])

---

Capabilities

SQL Capability

Query relational databases safely with validation.

from chatbot_framework.sql import SQLCapability, SQLConnection

# Connection is read-only
db = SQLConnection(
    connection_string="sqlite:///mydb.db",
    read_only=True,  # Enforced at connection level
)

sql = SQLCapability(connection=db)

# Usage in chatbot
chatbot = Chatbot(capabilities=[sql])
response = await chatbot.chat("What's our top product by revenue?")

Security Features:

• ✅ SELECT-only queries
• ✅ Read-only connection enforcement
• ✅ Query parameter validation
• ✅ Schema introspection
• ✅ Timeout protection

Web Capability

Fetch data from web resources with validation.

from chatbot_framework.web import WebCapability, SimpleWebClient

# Real or mock client
client = SimpleWebClient()  # Uses aiohttp
# or
from chatbot_framework.web import MockWebClient
client = MockWebClient()

web = WebCapability(client=client)

# Usage in chatbot
chatbot = Chatbot(capabilities=[web])
response = await chatbot.chat("What's the weather today?")

Features:

• ✅ HTTP GET/POST requests
• ✅ Timeout protection
• ✅ Mock client for testing
• ✅ Real client with aiohttp
• ✅ Domain whitelisting

RAG Capability

Retrieve and augment responses with document knowledge.

from chatbot_framework.rag import RAGCapability, VectorStore

# Setup vector store
vector_store = VectorStore()

# Load documents
vector_store.add_documents([
    Document(content="Our company values innovation..."),
    Document(content="Product A is for enterprise..."),
    Document(content="Pricing: $100/month..."),
])

rag = RAGCapability(vector_store=vector_store)

# Usage in chatbot
chatbot = Chatbot(capabilities=[rag])
response = await chatbot.chat("Tell me about our values")

Features:

• ✅ Document loading
• ✅ Semantic chunking
• ✅ Vector embeddings
• ✅ Similarity search
• ✅ Result ranking
• ✅ Source attribution

Custom Capabilities

Implement the Capability protocol to add custom functionality.

from chatbot_framework.core import Capability
from typing import Any

class CustomCapability(Capability):
    def register(self, registry) -> None:
        """Register with the planner."""
        registry.register_tool("my_tool", self)
    
    def describe(self) -> dict[str, Any]:
        """Describe to the LLM."""
        return {
            "name": "my_tool",
            "description": "Does something special",
            "parameters": {...},
        }
    
    async def invoke(self, action: str, context: dict[str, Any]) -> Any:
        """Execute the capability."""
        # Your implementation
        return result

# Use it
custom = CustomCapability()
chatbot = Chatbot(capabilities=[custom])

---

How to Use

Basic Workflow

Step 1: Setup

import asyncio
from chatbot_framework import Chatbot
from chatbot_framework.config import configure_production

async def main():
    # Configure framework
    configure_production()
    
    # Create chatbot
    chatbot = Chatbot(capabilities=[...])

Step 2: Chat

    # Single-turn chat
    response = await chatbot.chat("What's in our database?")
    print(f"Answer: {response.text}")
    print(f"Reasoning: {response.reasoning}")

Step 3: Handle Response

    # Response object has:
    # - text: The answer
    # - reasoning: How it was derived
    # - sources: What was used
    # - success: Whether it succeeded
    # - error: Error message if failed

asyncio.run(main())

Multi-Turn Conversations

async def main():
    chatbot = Chatbot(capabilities=[...])
    
    # Turn 1
    response1 = await chatbot.chat("What's our revenue?")
    print(response1.text)
    
    # Turn 2 (context is maintained)
    response2 = await chatbot.chat("Compare with last year")
    print(response2.text)
    
    # Turn 3
    response3 = await chatbot.chat("What about competitors?")
    print(response3.text)

With Configuration

from chatbot_framework.config import FrameworkConfig

async def main():
    # Custom configuration
    config = FrameworkConfig(
        debug=False,
        environment="production",
        limits={
            "max_planning_steps": 15,
            "max_replans": 5,
            "max_execution_time_seconds": 120,
        },
        security={
            "enable_tool_authorization": True,
            "allowed_domains": ["api.company.com", "data.company.com"],
        },
    )
    set_config(config)
    
    chatbot = Chatbot(capabilities=[...])
    response = await chatbot.chat("Your question")

With Observability

from chatbot_framework.observability import (
    EventEmitter,
    LoggingEventSink,
    get_emitter,
    set_emitter,
)
import logging

async def main():
    # Setup logging
    logging.basicConfig(level=logging.DEBUG)
    
    # Setup observability
    sink = LoggingEventSink("my_app")
    emitter = EventEmitter(sink=sink)
    set_emitter(emitter)
    
    chatbot = Chatbot(capabilities=[...])
    response = await chatbot.chat("Your question")
    
    # All events logged automatically

With Memory

from chatbot_framework.memory import MemoryManager

async def main():
    memory = MemoryManager()
    chatbot = Chatbot(capabilities=[...])
    
    # Turn 1
    memory.add_conversation_turn("user", "Hello")
    response1 = await chatbot.chat("Hello")
    memory.add_conversation_turn("assistant", response1.text)
    
    # Turn 2 - can access conversation history
    history = memory.get_conversation_context(num_turns=5)
    response2 = await chatbot.chat("Continue from before")
    memory.add_conversation_turn("assistant", response2.text)
    
    # Store facts
    memory.store_fact("user_name", "Alice")
    memory.learn_pattern("preference", "detailed_answers")

---

Examples

The framework includes 5 production-ready example applications demonstrating different capability levels:

Example 1: Minimal Chatbot

File: examples/01_minimal_chatbot.py

Text-only chatbot with no external dependencies.

python examples/01_minimal_chatbot.py

Features:

• No external capabilities
• Simple rule-based responses
• Zero dependencies
• Good starting point

Example 2: Knowledge Chatbot

File: examples/02_knowledge_chatbot.py

RAG-based chatbot for knowledge base queries.

python examples/02_knowledge_chatbot.py

Features:

• RAG retrieval from documents
• Grounded answers
• Topic-based routing
• Document search

Example 3: Enterprise Chatbot

File: examples/03_enterprise_chatbot.py

Business intelligence with SQL and Web capabilities.

python examples/03_enterprise_chatbot.py

Features:

• Database queries
• Web data fetching
• Multi-capability routing
• Business metrics

Example 4: Research Chatbot

File: examples/04_research_chatbot.py

Multi-source research assistant with reflection.

python examples/04_research_chatbot.py

Features:

• RAG + Web + Reflection
• Deep analysis
• Citation tracking
• Multi-source synthesis

Example 5: Advanced Agent

File: examples/05_advanced_agent.py

Full-featured agent with all capabilities.

python examples/05_advanced_agent.py

Features:

• All capabilities enabled
• Conversation memory
• Execution statistics
• Auto-capability selection
• Deep reasoning

---

API Reference

Main Classes

Chatbot

Main interface to the framework.

class Chatbot:
    def __init__(
        self,
        capabilities: list[Capability] | None = None,
        max_steps: int = 10,
        max_replans: int = 3,
        max_execution_time: int = 60,
    ) -> None:
        """Initialize chatbot."""
    
    async def chat(self, user_message: str) -> ChatResponse:
        """Process user message and return response."""

Properties:

• capabilities: List of enabled capabilities
• max_steps: Maximum planning steps
• max_replans: Maximum replan attempts
• max_execution_time: Timeout in seconds

ChatResponse

Response from the chatbot.

class ChatResponse:
    text: str                    # Main response text
    reasoning: str               # How it was derived
    sources: list[str]          # Sources used
    success: bool               # Whether successful
    error: str | None           # Error message if failed
    execution_time_ms: float    # Time taken

MemoryManager

Manages conversation and long-term memory.

class MemoryManager:
    def add_conversation_turn(self, role: str, content: str) -> None:
        """Add a conversation turn."""
    
    def get_conversation_context(self, num_turns: int = 5) -> str:
        """Get recent conversation context."""
    
    def store_fact(self, key: str, value: str) -> None:
        """Store a fact."""
    
    def recall_fact(self, key: str) -> str | None:
        """Recall a stored fact."""
    
    def learn_pattern(self, pattern_type: str, pattern: str) -> None:
        """Learn a pattern."""

CompositeValidator

Aggregates validation checks.

class CompositeValidator:
    def __init__(self, validators: list[Validator]) -> None:
        """Initialize with validators."""
    
    def validate(self, data: dict[str, Any]) -> ValidationResult:
        """Validate data against all validators."""

FrameworkConfig

Configuration management.

class FrameworkConfig:
    limits: LimitConfig
    security: SecurityConfig
    debug: bool
    environment: str
    
    @classmethod
    def development(cls) -> "FrameworkConfig":
        """Create development config."""
    
    @classmethod
    def production(cls) -> "FrameworkConfig":
        """Create production config."""

EventEmitter

Emit observability events.

class EventEmitter:
    def __init__(self, sink: EventSink | None = None) -> None:
        """Initialize emitter."""
    
    async def emit(
        self,
        event_type: EventType,
        component: str,
        message: str = "",
        duration_ms: float | None = None,
        metadata: dict[str, Any] | None = None,
        error: str | None = None,
        success: bool = True,
    ) -> None:
        """Emit an event."""

Models

from chatbot_framework.models import (
    Plan,
    Step,
    ChatRequest,
    ChatResponse,
    ExecutionContext,
    ExecutionState,
    ValidationResult,
    ConversationMemory,
    ApplicationState,
    LongTermMemory,
)

---

Configuration & Security

Configuration

Configure the framework for your environment:

from chatbot_framework.config import (
    configure_development,
    configure_production,
    get_config,
    set_config,
)

# Quick presets
configure_development()  # Permissive, debug enabled
configure_production()   # Strict, debug disabled

# Custom configuration
from chatbot_framework.config import FrameworkConfig

config = FrameworkConfig.production()
config.limits.max_planning_steps = 20
config.security.blocked_tools = ["dangerous_tool"]
set_config(config)

Security Features

Credential Detection

config = get_config()
security = config.security

# Automatic detection of sensitive keys
if security.is_credential_key("api_key"):
    print("This is a credential!")

# Filter sensitive data
data = {
    "username": "alice",
    "password": "secret123",
}
filtered = security.filter_sensitive_data(data)
# → {"username": "alice", "password": "[REDACTED]"}

Authorization

# Check if tool is allowed
if config.security.is_tool_allowed("fetch_data"):
    # Safe to use this tool
    pass

# Check if domain is allowed
if config.security.is_domain_allowed("api.company.com"):
    # Safe to fetch from this domain
    pass

Centralized Limits

limits = config.limits

# Validate against limits
if limits.validate_step_count(current_steps):
    # Can execute more steps
    pass

if limits.validate_replan_count(current_replans):
    # Can replan
    pass

if limits.validate_tool_calls(current_calls):
    # Can call more tools
    pass

---

Observability

Event Types

The framework emits structured events for all major operations:

from chatbot_framework.observability import EventType

# Execution events
EventType.EXECUTION_START
EventType.EXECUTION_COMPLETE
EventType.EXECUTION_ERROR

# Planning events
EventType.PLAN_START
EventType.PLAN_COMPLETE

# Tool events
EventType.TOOL_CALL_START
EventType.TOOL_CALL_COMPLETE
EventType.TOOL_CALL_ERROR

# Validation events
EventType.VALIDATION_START
EventType.VALIDATION_COMPLETE
EventType.VALIDATION_FAILED

# Reflection events
EventType.REFLECTION_START
EventType.REFLECTION_COMPLETE
EventType.REFLECTION_RETRY

# Memory events
EventType.MEMORY_STORE
EventType.MEMORY_RETRIEVE

# Capability events
EventType.CAPABILITY_REGISTER
EventType.CAPABILITY_INVOKE

Using Observability

from chatbot_framework.observability import (
    EventEmitter,
    LoggingEventSink,
    emit_event,
    get_emitter,
    set_emitter,
    EventType,
)
import logging

# Setup logging
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Create and set emitter
sink = LoggingEventSink("chatbot_framework")
emitter = EventEmitter(sink=sink)
set_emitter(emitter)

# Emit events manually
async def my_operation():
    await emit_event(
        event_type=EventType.EXECUTION_START,
        component="my_component",
        message="Starting operation",
    )
    
    # ... do work ...
    
    await emit_event(
        event_type=EventType.EXECUTION_COMPLETE,
        component="my_component",
        duration_ms=123.5,
        message="Operation completed",
    )

Custom Event Sinks

Implement custom event handling:

from chatbot_framework.observability import EventEmitter, EventSink
from chatbot_framework.observability.events import ObservabilityEvent

class CustomEventSink(EventSink):
    async def emit(self, event: ObservabilityEvent) -> None:
        # Send to your monitoring service
        await send_to_datadog(event.to_log_string())
        # Or store in database
        await db.insert_event(event)
        # Or send to webhook
        await requests.post("https://monitoring.com/events", json=event.dict())

# Use custom sink
custom_sink = CustomEventSink()
emitter = EventEmitter(sink=custom_sink)
set_emitter(emitter)

---

Advanced Usage

Multi-Capability Orchestration

The planner automatically identifies which capabilities to use:

chatbot = Chatbot(
    capabilities=[
        SQLCapability(...),
        WebCapability(...),
        RAGCapability(...),
    ]
)

# Single query triggers multi-step plan
response = await chatbot.chat(
    "Compare our sales with industry benchmarks "
    "and explain from company docs"
)

# Internally generates:
# Step 1: SELECT revenue FROM sales (SQL)
# Step 2: GET https://api.industry.com/benchmarks (Web)
# Step 3: Search company docs for explanation (RAG)
# Step 4: Synthesize answer from all sources

Custom Tool Implementation

Extend the framework with custom tools:

from chatbot_framework.tools import Tool

class CustomTool(Tool):
    def __init__(self):
        super().__init__(
            name="my_tool",
            description="Does something unique",
        )
    
    async def execute(self, **kwargs) -> str:
        # Your implementation
        result = await process(**kwargs)
        return result

# Register and use
tool = CustomTool()
chatbot.register_tool(tool)

Streaming Responses

Use async generators for streaming:

async def stream_response(chatbot, query):
    async for event in chatbot.stream_chat(query):
        if hasattr(event, 'text'):
            yield event.text  # Stream partial response
        elif hasattr(event, 'error'):
            yield f"Error: {event.error}"

Batch Processing

Process multiple queries efficiently:

async def batch_chat(chatbot, queries):
    import asyncio
    
    # Process in parallel
    tasks = [chatbot.chat(q) for q in queries]
    responses = await asyncio.gather(*tasks)
    return responses

# Usage
queries = ["Query 1", "Query 2", "Query 3"]
responses = await batch_chat(chatbot, queries)

Fallback Mechanisms

Graceful degradation when capabilities fail:

from chatbot_framework import Chatbot

async def robust_chat(chatbot, query):
    try:
        # Try with all capabilities
        response = await chatbot.chat(query)
        if response.success:
            return response
    except Exception as e:
        print(f"Full execution failed: {e}")
    
    # Fallback: try with fewer capabilities
    try:
        # Disable web capability and retry
        chatbot.capabilities = [
            c for c in chatbot.capabilities 
            if c.__class__.__name__ != 'WebCapability'
        ]
        response = await chatbot.chat(query)
        return response
    except Exception as e:
        print(f"Fallback also failed: {e}")
        return None

---

Best Practices

1. Always Use Production Config

from chatbot_framework import configure_production

# In your application
configure_production()  # Strict limits, security enabled

2. Implement Error Handling

response = await chatbot.chat("Query")
if not response.success:
    # Handle gracefully
    logger.error(f"Chat failed: {response.error}")
    return "I encountered an error. Please try again."

3. Monitor with Observability

# Setup logging
import logging
logging.basicConfig(level=logging.INFO)

# Enable observability
from chatbot_framework.observability import emit_event, EventType

# Events automatically emitted for all operations

4. Use Validators

from chatbot_framework.validation import CompositeValidator

validator = CompositeValidator([...])
result = validator.validate(response.text)
if not result.is_valid():
    logger.warning(f"Output validation failed: {result.message}")

5. Manage Memory Explicitly

memory = MemoryManager()

# Add every turn
memory.add_conversation_turn("user", user_input)
memory.add_conversation_turn("assistant", response.text)

# Periodically export for backup
state = memory.export_state()
save_to_db(state)

6. Test with Examples

Start with example applications to understand patterns:

# Run examples to learn
python examples/01_minimal_chatbot.py
python examples/02_knowledge_chatbot.py
python examples/03_enterprise_chatbot.py

7. Use Async Properly

# Don't block event loop
response = await chatbot.chat(query)  # ✓ Correct

# Not this
import time
time.sleep(1)  # ✗ Blocks entire loop

# Use asyncio.sleep instead
await asyncio.sleep(1)  # ✓ Correct

8. Validate Input

# Validate user input before processing
if not user_query or len(user_query) > 10000:
    return "Invalid query"

# Sanitize for security
user_query = user_query.strip()

response = await chatbot.chat(user_query)

---

Troubleshooting

"Module not found" errors

Problem: ImportError: No module named 'anthropic'

Solution: Install the optional dependencies:

pip install chatbot-framework[openai]
pip install chatbot-framework[qdrant]
pip install chatbot-framework[sql]

Type checking errors with Pyright

Problem: reportUnknownMemberType or reportUnknownArgumentType

Solution: Ensure you have all type hints:

# Instead of
result = some_function()

# Use
result: dict[str, Any] = some_function()

Tests failing

Problem: Test failures after installation

Solution: Ensure dev dependencies are installed:

pip install -e ".[dev]"
pytest tests/ -v

Chatbot not generating plans

Problem: Chatbot.chat() returns empty plans

Solution: Check the LLM is configured:

from chatbot_framework import Chatbot

# Must have LLM available
chatbot = Chatbot(capabilities=[...])  # Ensure LLM is set up

Memory not persisting

Problem: Conversation history lost between runs

Solution: Save memory state:

memory = MemoryManager()
# ... chat turns ...

# Export before shutdown
state = memory.export_state()
save_to_file(state)

# Restore on startup
memory.import_state(state)

Security warnings in logs

Problem: Seeing [REDACTED] in logs

Solution: This is expected! Credentials are automatically filtered. Check the security config:

from chatbot_framework.config import get_config

config = get_config()
print(config.security.credential_patterns)

Performance issues

Problem: Chatbot is slow

Solution: Check limits and timeouts:

from chatbot_framework.config import get_config

config = get_config()
print(f"Max steps: {config.limits.max_planning_steps}")
print(f"Timeout: {config.limits.max_execution_time_seconds}s")

# Reduce if needed
config.limits.max_planning_steps = 5

Unable to connect to database

Problem: SQLConnection failed to connect

Solution: Verify connection string and permissions:

from chatbot_framework.sql import SQLConnection

# Test connection
try:
    db = SQLConnection("sqlite:///test.db", read_only=True)
    print("✓ Connected")
except Exception as e:
    print(f"✗ Failed: {e}")

---

Integration with Existing Applications

Building a Chatbot with Your SQL Database and Vector Store

If you already have:

• ✅ A SQL database (PostgreSQL, MySQL, SQLite, etc.)
• ✅ A vector database (Qdrant, Weaviate, Pinecone, etc.)
• ✅ Internal documents/knowledge base

Here's how easy it is to add an AI chatbot:

Super Quick (10 lines)

import asyncio
from chatbot_framework import Chatbot
from chatbot_framework.llm.anthropic_adapter import AnthropicLLM
from chatbot_framework.sql import SQLCapability, SQLConnection
from chatbot_framework.rag import RAGCapability, VectorStore

async def main():
    # 1. LLM
    llm = AnthropicLLM()
    
    # 2. Connect to your existing SQL database
    db = SQLConnection("postgresql://user:pass@localhost/mydb", read_only=True)
    sql = SQLCapability(connection=db)
    
    # 3. Connect to your existing vector database
    rag_store = VectorStore(host="localhost", port=6333)
    rag = RAGCapability(vector_store=rag_store)
    
    # 4. Create chatbot
    bot = Chatbot(llm=llm, capabilities=[sql, rag])
    
    # 5. Use it!
    response = await bot.achat("Compare our sales with industry data from docs")
    print(response.text)

asyncio.run(main())

That's it! 5 steps, minimal code.

What Happens Automatically

When you ask the chatbot a question, the framework:

1. Planner analyzes query
   ├─ Needs SQL data? → Yes
   ├─ Needs document data? → Yes
   └─ Strategy: Compare and explain

2. Executor creates multi-step plan
   ├─ Step 1: Query your SQL database
   ├─ Step 2: Search your vector database
   └─ Step 3: Synthesize results

3. Results combined intelligently
   ├─ Pulls from your SQL
   ├─ Pulls from your documents
   └─ LLM creates unified answer

4. Response returned to user

Common Integration Patterns

Pattern 1: Add to FastAPI Application

from fastapi import FastAPI
from chatbot_framework import Chatbot

app = FastAPI()
chatbot = None

@app.on_event("startup")
async def startup():
    global chatbot
    chatbot = await create_chatbot()

@app.post("/chat")
async def chat(message: str):
    response = await chatbot.achat(message)
    return {"answer": response.text}

Pattern 2: Add to Flask Application

from flask import Flask, request
from chatbot_framework import Chatbot
import asyncio

app = Flask(__name__)
chatbot = None

def init_chatbot():
    global chatbot
    chatbot = asyncio.run(create_chatbot())

@app.route("/chat", methods=["POST"])
def chat():
    message = request.json["message"]
    response = asyncio.run(chatbot.achat(message))
    return {"answer": response.text}

if __name__ == "__main__":
    init_chatbot()
    app.run()

Pattern 3: Add to Existing CLI Tool

import asyncio
from chatbot_framework import Chatbot

async def main():
    chatbot = await create_chatbot()
    
    while True:
        query = input("You: ")
        response = await chatbot.achat(query)
        print(f"Bot: {response.text}\n")

if __name__ == "__main__":
    asyncio.run(main())

Pattern 4: Add to Background Job

import asyncio
from celery import Celery
from chatbot_framework import Chatbot

app = Celery()
chatbot = None

@app.task
def process_chat_request(user_id, query):
    response = asyncio.run(chatbot.achat(query))
    
    # Save response
    save_to_database(user_id, query, response.text)
    
    # Send notification
    send_email(user_id, response.text)
    
    return response

def init_chatbot():
    global chatbot
    chatbot = asyncio.run(create_chatbot())

if __name__ == "__main__":
    init_chatbot()
    app.worker_main()

Using Your Existing Vector Database

from chatbot_framework.rag import RAGCapability, VectorStore

# Connect to your Qdrant instance
vector_store = VectorStore(
    host="your-qdrant-server.com",
    port=6333,
    collection_name="your_existing_collection",
)

# Your documents are already indexed - just use them!
rag = RAGCapability(vector_store=vector_store)

# Now queries automatically search your existing documents

Using Your Existing SQL Database

from chatbot_framework.sql import SQLCapability, SQLConnection

# Connect to your existing database
db = SQLConnection(
    connection_string="postgresql://user:pass@prod.example.com/company_db",
    read_only=True,  # Safety first!
)

sql = SQLCapability(connection=db)

# Now queries automatically run against your existing schema
# The framework learns your schema and generates safe queries

Cost & Performance

Minimal overhead:

• No additional infrastructure needed
• Works with existing databases
• Minimal latency (~1-2s per query)
• Scales with your infrastructure

Example costs (with Anthropic Claude):

• Simple question: ~$0.001
• Multi-source question: ~$0.005
• Batch of 100 questions: ~$0.30

What Makes It Easy

Aspect	Benefit
One API	Single chatbot.achat() call for all queries
Automatic Routing	Framework decides which capabilities to use
No Schema Learning	Framework introspects your database
No Prompt Engineering	Framework handles all LLM prompting
Type-Safe	Full type hints, IDE autocomplete
Observable	Built-in logging and event tracking
Tested	232 tests, production-ready

Example Queries Your Chatbot Handles

"How many employees joined this year?"
→ Routes to SQL automatically

"What's our leave policy?"
→ Routes to RAG (your documents) automatically

"Compare our sales with industry benchmarks and explain using our docs"
→ Routes to SQL + RAG + Web automatically
→ Synthesizes results
→ Generates unified answer

"Create a report on Q2 performance"
→ Could route to SQL + RAG + Web
→ Formats results into report
→ Returns structured output

Real-World Example

See examples/06_sql_and_rag_chatbot.py for complete production example with:

• SQL setup
• RAG setup
• Multi-query examples
• Production configuration
• Error handling

See examples/06_quickstart_sql_rag.py for minimal 10-line example.

---

Development

Running Tests

# All tests
pytest tests/ -v

# Specific test file
pytest tests/unit/test_config.py -v

# Specific test
pytest tests/unit/test_config.py::TestLimitConfig::test_default_limits -v

# With coverage
pytest tests/ --cov=src/chatbot_framework

Type Checking

# Check all code
pyright src/

# With strict mode
pyright src/ --level=error

# Check specific file
pyright src/chatbot_framework/core.py

Linting

# Check
ruff check src/

# Fix automatically
ruff check src/ --fix

# Check specific rules
ruff check src/ --select E,W,F

Code Quality

# All checks
pytest tests/ -v                              # Tests
pyright src/ --level=error                   # Type checking
ruff check src/ --fix                        # Linting

---

Additional Resources

• Tests: tests/ directory contains 232 tests organized in 4 tiers
• Examples: examples/ directory has 5 production examples
• Documentation: PHASE10.md for detailed Phase 10 features
• Source: src/chatbot_framework/ organized by feature

---

Summary

The Chatbot Framework is a sophisticated, production-ready system for building AI chatbots. It provides:

✅ Modular Architecture: Use only what you need ✅ Type Safety: Full type annotations, strict mode compatible ✅ Security: Credential filtering, authorization, centralized limits ✅ Observability: Structured events, pluggable sinks ✅ Quality Control: Validation gates, reflection, memory ✅ Examples: 5 production-ready applications ✅ Testing: 232 comprehensive tests ✅ Documentation: Complete guides and API reference

Get started with the Quick Start section, explore the Examples, and refer to the API Reference as needed.

Happy chatbot building! 🚀 ruff check src/ tests/

## License

MIT