feat: Deploy Physical AI & Humanoid Robotics RAG backend

- Add RAG backend with Cohere and Qdrant integration
- Include API endpoints for chat functionality
- Configure Dockerfile for Hugging Face Spaces deployment
- Add proper health check endpoint
- Set up requirements and dependencies

.env.example

@@ -0,0 +1,26 @@
+# OpenRouter API Configuration
+OPENROUTER_API_KEY=your_openrouter_api_key_here
+
+# Qdrant Vector Database Configuration
+QDRANT_URL=your_qdrant_url_here
+QDRANT_API_KEY=your_qdrant_api_key_here
+QDRANT_CLUSTER_ID=your_qdrant_cluster_id_here
+
+# Neon PostgreSQL Database Configuration
+NEON_DATABASE_URL=your_neon_database_url_here
+
+# Cohere API Key (if needed)
+COHERE_API_KEY=your_cohere_api_key_here
+
+# Backend API Key
+BACKEND_API_KEY=your_backend_api_key_here
+
+# Target URL for Docusaurus site
+TARGET_URL=your_vercel_url_here
+
+# Application Configuration
+DEBUG=False
+LOG_LEVEL=INFO
+MAX_CONTENT_LENGTH=5000
+RATE_LIMIT_REQUESTS=100
+RATE_LIMIT_WINDOW=60

.gitignore

@@ -0,0 +1,26 @@
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+.Python
+env/
+venv/
+.venv/
+pip-log.txt
+pip-delete-this-directory.txt
+.tox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.log
+.git/
+.DS_Store
+.DS_Store?
+._*
+.Spotlight-V100
+.Trashes
+ehthumbs.db
+Thumbs.db

Dockerfile

@@ -0,0 +1,44 @@
+# Use Python 3.11 slim image as base
+FROM python:3.11-slim
+
+# Set environment variables
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PYTHONPATH=/app \
+    PORT=7860
+
+# Set work directory
+WORKDIR /app
+
+# Install system dependencies
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends \
+        build-essential \
+        gcc \
+        curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements first to leverage Docker cache
+COPY requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir --upgrade pip \
+    && pip install --no-cache-dir -r requirements.txt
+
+# Copy the rest of the application
+COPY . .
+
+# Create a non-root user and set permissions
+RUN adduser --disabled-password --gecos '' appuser \
+    && chown -R appuser:appuser /app
+USER appuser
+
+# Expose port (Hugging Face typically uses port 7860 or 8080)
+EXPOSE $PORT
+
+# Health check endpoint
+HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
+    CMD curl -f http://localhost:$PORT/health || exit 1
+
+# Run the application with uvicorn directly for better production performance
+CMD ["sh", "-c", "python app.py"]

README.md

@@ -1,10 +1,76 @@
----
-title: Chat Bot
-emoji: 🌖
-colorFrom: pink
-colorTo: red
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Docusaurus Embedding Pipeline
+
+This project extracts text from deployed Docusaurus URLs, generates embeddings using Cohere, and stores them in Qdrant for RAG-based retrieval.
+
+## Features
+
+- Crawls Docusaurus sites to extract all accessible URLs
+- Extracts and cleans text content from each page
+- Chunks large documents to optimize embedding quality
+- Generates vector embeddings using Cohere's API
+- Stores embeddings in Qdrant vector database with metadata
+- Supports similarity search for RAG applications
+
+## Prerequisites
+
+- Python 3.9+
+- UV package manager (`pip install uv`)
+- Cohere API key
+- Qdrant instance (local or cloud)
+
+## Setup
+
+1. Clone the repository and navigate to the backend directory
+2. Install UV package manager:
+   ```bash
+   pip install uv
+   ```
+
+3. Install dependencies:
+   ```bash
+   cd backend
+   uv sync  # or uv pip install -r requirements.txt
+   ```
+
+4. Set up environment variables:
+   ```bash
+   cp .env.example .env
+   # Edit .env with your Cohere API key and Qdrant configuration
+   ```
+
+## Configuration
+
+The pipeline can be configured via environment variables in the `.env` file:
+
+- `COHERE_API_KEY`: Your Cohere API key
+- `QDRANT_URL`: URL to your Qdrant instance
+- `QDRANT_API_KEY`: API key for Qdrant (if required)
+- `TARGET_URL`: The Docusaurus site to process
+
+## Usage
+
+Run the complete pipeline:
+```bash
+uv run main.py
+```
+
+## Architecture
+
+The pipeline consists of these main functions:
+
+1. `get_all_urls()` - Extracts all URLs from the target Docusaurus site
+2. `extract_text_from_url()` - Cleans and extracts text content from a URL
+3. `chunk_text()` - Splits large documents into manageable chunks
+4. `embed()` - Generates vector embeddings using Cohere
+5. `create_collection()` - Sets up the Qdrant collection
+6. `save_chunk_to_qdrant()` - Stores embeddings with metadata in Qdrant
+
+The main function orchestrates the complete workflow from crawling to storage.
+
+## Output
+
+The pipeline stores document chunks as vectors in a Qdrant collection named "rag_embedding" with the following metadata:
+- Content text
+- Source URL
+- Position in original document
+- Creation timestamp

agent.py

@@ -0,0 +1,254 @@
+import os
+import json
+import logging
+from typing import Dict, List, Any
+from dotenv import load_dotenv
+import asyncio
+import time
+
+# Load environment variables
+load_dotenv()
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+def retrieve_information(query: str, top_k: int = 5, threshold: float = 0.3) -> Dict:
+    """
+    Retrieve information from the knowledge base based on a query
+    """
+    from retrieving import RAGRetriever
+    retriever = RAGRetriever()
+
+    try:
+        # Call the existing retrieve method from the RAGRetriever instance
+        json_response = retriever.retrieve(query_text=query, top_k=top_k, threshold=threshold)
+        results = json.loads(json_response)
+
+        # Format the results for the assistant
+        formatted_results = []
+        for result in results.get('results', []):
+            formatted_results.append({
+                'content': result['content'],
+                'url': result['url'],
+                'position': result['position'],
+                'similarity_score': result['similarity_score'],
+                'chunk_id': result.get('chunk_id', ''),
+                'created_at': result.get('created_at', '')
+            })
+
+        return {
+            'query': query,
+            'retrieved_chunks': formatted_results,
+            'total_results': len(formatted_results),
+            'metadata': results.get('metadata', {})
+        }
+    except Exception as e:
+        logger.error(f"Error in retrieve_information: {e}")
+        return {
+            'query': query,
+            'retrieved_chunks': [],
+            'total_results': 0,
+            'error': str(e),
+            'metadata': {}
+        }
+
+class RAGAgent:
+    def __init__(self):
+        # Initialize the RAG system components
+        # For now, we'll use the retrieval function directly
+        # In a real implementation, you would initialize your existing RAG components
+        logger.info("RAG Agent initialized with retrieval and generation components")
+
+    def query_agent(self, query_text: str, session_id: str = None, query_type: str = "global", selected_text: str = None) -> Dict:
+        """
+        Process a query through the RAG system and return structured response
+        """
+        start_time = time.time()
+
+        logger.info(f"Processing query through RAG system: '{query_text[:50]}...'")
+
+        try:
+            # Retrieve relevant information using our retrieval system
+            retrieval_result = retrieve_information(query_text, top_k=5, threshold=0.3)
+
+            if retrieval_result.get('error'):
+                return {
+                    "answer": "Sorry, I encountered an error retrieving information.",
+                    "sources": [],
+                    "matched_chunks": [],
+                    "citations": [],
+                    "error": retrieval_result['error'],
+                    "query_time_ms": (time.time() - start_time) * 1000,
+                    "session_id": session_id,
+                    "query_type": query_type
+                }
+
+            # Format the retrieved information for response generation
+            # In a real implementation, you would connect this to your response generator
+            retrieved_chunks = retrieval_result.get('retrieved_chunks', [])
+
+            if not retrieved_chunks:
+                return {
+                    "answer": "I couldn't find relevant information in the Physical AI & Humanoid Robotics curriculum to answer your question. Please try asking about specific topics from the curriculum like ROS 2, Digital Twins, AI-Brain, or VLA.",
+                    "sources": [],
+                    "matched_chunks": [],
+                    "citations": [],
+                    "error": None,
+                    "query_time_ms": (time.time() - start_time) * 1000,
+                    "session_id": session_id,
+                    "query_type": query_type
+                }
+
+            # Generate a response based on the retrieved information
+            # For now, we'll create a simple response based on the retrieved chunks
+            answer_parts = ["Based on the Physical AI & Humanoid Robotics curriculum:"]
+
+            # Include content from the most relevant chunks
+            for i, chunk in enumerate(retrieved_chunks[:2]):  # Use top 2 chunks
+                content = chunk.get('content', '')[:300]  # Limit content length
+                answer_parts.append(f"{content}...")
+
+            answer = " ".join(answer_parts)
+
+            # Create citations from the retrieved chunks
+            citations = []
+            for chunk in retrieved_chunks:
+                citation = {
+                    "document_id": chunk.get('chunk_id', ''),
+                    "title": chunk.get('url', ''),
+                    "chapter": "",
+                    "section": "",
+                    "page_reference": ""
+                }
+                citations.append(citation)
+
+            # Calculate query time
+            query_time_ms = (time.time() - start_time) * 1000
+
+            # Format the response
+            response = {
+                "answer": answer,
+                "sources": [chunk.get('url', '') for chunk in retrieved_chunks if chunk.get('url')],
+                "matched_chunks": retrieved_chunks,
+                "citations": citations,
+                "query_time_ms": query_time_ms,
+                "session_id": session_id,
+                "query_type": query_type,
+                "confidence": self._calculate_confidence(retrieved_chunks),
+                "error": None
+            }
+
+            logger.info(f"Query processed in {query_time_ms:.2f}ms")
+            return response
+
+        except Exception as e:
+            logger.error(f"Error processing query: {e}")
+            return {
+                "answer": "Sorry, I encountered an error processing your request.",
+                "sources": [],
+                "matched_chunks": [],
+                "citations": [],
+                "error": str(e),
+                "query_time_ms": (time.time() - start_time) * 1000,
+                "session_id": session_id,
+                "query_type": query_type
+            }
+
+    def _calculate_confidence(self, matched_chunks: List[Dict]) -> str:
+        """
+        Calculate confidence level based on similarity scores and number of matches
+        """
+        if not matched_chunks:
+            return "low"
+
+        avg_score = sum(chunk.get('similarity_score', 0.0) for chunk in matched_chunks) / len(matched_chunks)
+
+        if avg_score >= 0.7:
+            return "high"
+        elif avg_score >= 0.4:
+            return "medium"
+        else:
+            return "low"
+
+def query_agent(query_text: str) -> Dict:
+    """
+    Convenience function to query the RAG agent
+    """
+    agent = RAGAgent()
+    return agent.query_agent(query_text)
+
+def run_agent_sync(query_text: str) -> Dict:
+    """
+    Synchronous function to run the agent for direct usage
+    """
+    import asyncio
+
+    async def run_async():
+        agent = RAGAgent()
+        return await agent._async_query_agent(query_text)
+
+    # Check if there's already a running event loop
+    try:
+        loop = asyncio.get_running_loop()
+        # If there's already a loop, run in a separate thread
+        import concurrent.futures
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            future = executor.submit(asyncio.run, run_async())
+            return future.result()
+    except RuntimeError:
+        # No running loop, safe to use asyncio.run
+        return asyncio.run(run_async())
+
+def main():
+    """
+    Main function to demonstrate the RAG agent functionality
+    """
+    logger.info("Initializing RAG Agent...")
+
+    # Initialize the agent
+    agent = RAGAgent()
+
+    # Example queries to test the system
+    test_queries = [
+        "What is ROS2?",
+        "Explain humanoid design principles",
+        "How does VLA work?",
+        "What are simulation techniques?",
+        "Explain AI control systems"
+    ]
+
+    print("RAG Agent - Testing Queries")
+    print("=" * 50)
+
+    for i, query in enumerate(test_queries, 1):
+        print(f"\nQuery {i}: {query}")
+        print("-" * 30)
+
+        # Process query through agent
+        response = agent.query_agent(query)
+
+        # Print formatted results
+        print(f"Answer: {response['answer']}")
+
+        if response.get('sources'):
+            print(f"Sources: {len(response['sources'])} documents")
+            for source in response['sources'][:3]:  # Show first 3 sources
+                print(f"  - {source}")
+
+        if response.get('matched_chunks'):
+            print(f"Matched chunks: {len(response['matched_chunks'])}")
+            for j, chunk in enumerate(response['matched_chunks'][:2], 1):  # Show first 2 chunks
+                content_preview = chunk['content'][:100] + "..." if len(chunk['content']) > 100 else chunk['content']
+                print(f"  Chunk {j}: {content_preview}")
+                print(f"    Source: {chunk['url']}")
+                print(f"    Score: {chunk['similarity_score']:.3f}")
+
+        print(f"Query time: {response['query_time_ms']:.2f}ms")
+        print(f"Confidence: {response.get('confidence', 'unknown')}")
+
+        if i < len(test_queries):  # Don't sleep after the last query
+            time.sleep(1)  # Small delay between queries
+
+if __name__ == "__main__":
+    main()

api.py

@@ -0,0 +1,215 @@
+import os
+import asyncio
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+from typing import List, Optional, Dict
+from dotenv import load_dotenv
+import logging
+
+# Load environment variables
+load_dotenv()
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Import the existing RAG agent functionality
+from agent import RAGAgent
+
+# Create FastAPI app
+app = FastAPI(
+    title="RAG Agent API",
+    description="API for RAG Agent with document retrieval and question answering",
+    version="1.0.0"
+)
+
+# Add CORS middleware for development
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # In production, replace with specific origins
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Pydantic models
+class QueryRequest(BaseModel):
+    query: str
+
+class ChatRequest(BaseModel):
+    query: str
+    message: str
+    session_id: str
+    selected_text: Optional[str] = None
+    query_type: str = "global"
+    top_k: int = 5
+
+class MatchedChunk(BaseModel):
+    content: str
+    url: str
+    position: int
+    similarity_score: float
+
+class QueryResponse(BaseModel):
+    answer: str
+    sources: List[str]
+    matched_chunks: List[MatchedChunk]
+    error: Optional[str] = None
+    status: str  # "success", "error", "empty"
+    query_time_ms: Optional[float] = None
+    confidence: Optional[str] = None
+
+class ChatResponse(BaseModel):
+    response: str
+    citations: List[Dict[str, str]]
+    session_id: str
+    query_type: str
+    timestamp: str
+
+class HealthResponse(BaseModel):
+    status: str
+    message: str
+
+# Global RAG agent instance
+rag_agent = None
+
+@app.on_event("startup")
+async def startup_event():
+    """Initialize the RAG agent on startup"""
+    global rag_agent
+    logger.info("Initializing RAG Agent...")
+    try:
+        rag_agent = RAGAgent()
+        logger.info("RAG Agent initialized successfully")
+    except Exception as e:
+        logger.error(f"Failed to initialize RAG Agent: {e}")
+        raise
+
+@app.post("/ask", response_model=QueryResponse)
+async def ask_rag(request: QueryRequest):
+    """
+    Process a user query through the RAG agent and return the response
+    """
+    logger.info(f"Processing query: {request.query[:50]}...")
+
+    try:
+        # Validate input
+        if not request.query or len(request.query.strip()) == 0:
+            raise HTTPException(status_code=400, detail="Query cannot be empty")
+
+        if len(request.query) > 2000:
+            raise HTTPException(status_code=400, detail="Query too long, maximum 2000 characters")
+
+        # Process query through RAG agent
+        response = rag_agent.query_agent(request.query)
+
+        # Format response
+        formatted_response = QueryResponse(
+            answer=response.get("answer", ""),
+            sources=response.get("sources", []),
+            matched_chunks=[
+                MatchedChunk(
+                    content=chunk.get("content", ""),
+                    url=chunk.get("url", ""),
+                    position=chunk.get("position", 0),
+                    similarity_score=chunk.get("similarity_score", 0.0)
+                )
+                for chunk in response.get("matched_chunks", [])
+            ],
+            error=response.get("error"),
+            status="error" if response.get("error") else "success",
+            query_time_ms=response.get("query_time_ms"),
+            confidence=response.get("confidence")
+        )
+
+        logger.info(f"Query processed successfully in {response.get('query_time_ms', 0):.2f}ms")
+        return formatted_response
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error processing query: {e}")
+        return QueryResponse(
+            answer="",
+            sources=[],
+            matched_chunks=[],
+            error=str(e),
+            status="error"
+        )
+
+@app.post("/api", response_model=ChatResponse)
+async def chat_endpoint(request: ChatRequest):
+    """
+    Main chat endpoint that handles conversation with RAG capabilities
+    """
+    logger.info(f"Processing chat query: {request.query[:50]}...")
+
+    try:
+        # Validate input
+        if not request.query or len(request.query.strip()) == 0:
+            raise HTTPException(status_code=400, detail="Query cannot be empty")
+
+        if not request.session_id or len(request.session_id.strip()) == 0:
+            raise HTTPException(status_code=400, detail="Session ID cannot be empty")
+
+        if len(request.query) > 2000:
+            raise HTTPException(status_code=400, detail="Query too long, maximum 2000 characters")
+
+        # Process query through RAG agent
+        response = rag_agent.query_agent(request.query)
+
+        # Format response to match expected structure
+        from datetime import datetime
+        timestamp = datetime.utcnow().isoformat()
+
+        # Convert matched chunks to citations format
+        citations = []
+        for chunk in response.get("matched_chunks", []):
+            citation = {
+                "document_id": "",
+                "title": chunk.get("url", ""),
+                "chapter": "",
+                "section": "",
+                "page_reference": ""
+            }
+            citations.append(citation)
+
+        formatted_response = ChatResponse(
+            response=response.get("answer", ""),
+            citations=citations,
+            session_id=request.session_id,
+            query_type=request.query_type,
+            timestamp=timestamp
+        )
+
+        logger.info(f"Chat query processed successfully")
+        return formatted_response
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error processing chat query: {e}")
+        from datetime import datetime
+        return ChatResponse(
+            response="",
+            citations=[],
+            session_id=request.session_id,
+            query_type=request.query_type,
+            timestamp=datetime.utcnow().isoformat()
+        )
+
+@app.get("/health", response_model=HealthResponse)
+async def health_check():
+    """
+    Health check endpoint
+    """
+    return HealthResponse(
+        status="healthy",
+        message="RAG Agent API is running"
+    )
+
+# For running with uvicorn
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

app.py

@@ -0,0 +1,8 @@
+# app.py - Entry point for Hugging Face Spaces
+import os
+import uvicorn
+from api import app
+
+if __name__ == "__main__":
+    port = int(os.environ.get("PORT", 7860))
+    uvicorn.run(app, host="0.0.0.0", port=port)

main.py

@@ -0,0 +1,322 @@
+import os
+import requests
+from bs4 import BeautifulSoup
+import xml.etree.ElementTree as ET
+from typing import List, Dict, Any
+import cohere
+from qdrant_client import QdrantClient
+from qdrant_client.http import models
+from qdrant_client.models import PointStruct
+import logging
+from urllib.parse import urljoin, urlparse
+import time
+import uuid
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class DocusaurusEmbeddingPipeline:
+    def __init__(self):
+        # Initialize Cohere client
+        self.cohere_client = cohere.Client(api_key=os.getenv("COHERE_API_KEY"))
+
+        # Initialize Qdrant client
+        qdrant_url = os.getenv("QDRANT_URL", "http://localhost:6333")
+        qdrant_api_key = os.getenv("QDRANT_API_KEY")
+
+        if qdrant_api_key:
+            self.qdrant_client = QdrantClient(url=qdrant_url, api_key=qdrant_api_key)
+        else:
+            self.qdrant_client = QdrantClient(url=qdrant_url)
+
+        # Target URL for the Docusaurus site - configurable via environment variable
+        self.target_url = os.getenv("TARGET_URL", "https://your-vercel-url.vercel.app/")
+
+    def get_all_urls(self, base_url: str) -> List[str]:
+        """
+        Extract all URLs from a deployed Docusaurus site using sitemap
+        """
+        urls = []
+
+        try:
+            # Try to get URLs from sitemap first
+            sitemap_url = urljoin(base_url, "sitemap.xml")
+            response = requests.get(sitemap_url)
+
+            if response.status_code == 200:
+                root = ET.fromstring(response.content)
+
+                # Handle both sitemap index and regular sitemap
+                if root.tag.endswith('sitemapindex'):
+                    # If it's a sitemap index, get individual sitemaps
+                    for sitemap in root.findall('.//{http://www.sitemaps.org/schemas/sitemap/0.9}loc'):
+                        sitemap_response = requests.get(sitemap.text)
+                        if sitemap_response.status_code == 200:
+                            sitemap_root = ET.fromstring(sitemap_response.content)
+                            for url_elem in sitemap_root.findall('.//{http://www.sitemaps.org/schemas/sitemap/0.9}loc'):
+                                urls.append(url_elem.text)
+                else:
+                    # Regular sitemap
+                    for url_elem in root.findall('.//{http://www.sitemaps.org/schemas/sitemap/0.9}loc'):
+                        urls.append(url_elem.text)
+            else:
+                # Fallback: try to crawl the site by looking for links
+                logger.info(f"Sitemap not found at {sitemap_url}, attempting to crawl...")
+
+                # Get the main page and extract links
+                response = requests.get(base_url)
+                soup = BeautifulSoup(response.content, 'html.parser')
+
+                # Find all links within the page
+                for link in soup.find_all('a', href=True):
+                    href = link['href']
+                    full_url = urljoin(base_url, href)
+
+                    # Only add URLs from the same domain
+                    if urlparse(full_url).netloc == urlparse(base_url).netloc:
+                        if full_url not in urls and full_url.startswith(base_url):
+                            urls.append(full_url)
+
+        except Exception as e:
+            logger.error(f"Error getting URLs from {base_url}: {e}")
+
+        return urls
+
+    def extract_text_from_url(self, url: str) -> str:
+        """
+        Extract and clean text from a single URL
+        """
+        try:
+            response = requests.get(url)
+            response.raise_for_status()
+
+            soup = BeautifulSoup(response.content, 'html.parser')
+
+            # Remove script and style elements
+            for script in soup(["script", "style"]):
+                script.decompose()
+
+            # Look for main content containers typically used in Docusaurus
+            # Try multiple selectors to find the main content
+            content_selectors = [
+                'article',  # Main article content
+                '.markdown',  # Docusaurus markdown content
+                '.theme-doc-markdown',  # Docusaurus theme markdown
+                '.main-wrapper',  # Main content wrapper
+                'main',  # Main content area
+                '.container',  # Container with content
+                '[role="main"]'  # Main role
+            ]
+
+            content = ""
+            for selector in content_selectors:
+                elements = soup.select(selector)
+                if elements:
+                    for element in elements:
+                        # Get text but try to preserve some structure
+                        text = element.get_text(separator=' ', strip=True)
+                        if len(text) > len(content):
+                            content = text
+                    break
+
+            # If no specific content found, get all body text
+            if not content:
+                body = soup.find('body')
+                if body:
+                    content = body.get_text(separator=' ', strip=True)
+
+            # Clean up the text
+            lines = (line.strip() for line in content.splitlines())
+            chunks = (phrase.strip() for line in lines for phrase in line.split("  "))
+            content = ' '.join(chunk for chunk in chunks if chunk)
+
+            return content
+
+        except Exception as e:
+            logger.error(f"Error extracting text from {url}: {e}")
+            return ""
+
+    def chunk_text(self, text: str, chunk_size: int = 1000, overlap: int = 100) -> List[str]:
+        """
+        Split text into chunks with overlap to preserve context
+        """
+        if len(text) <= chunk_size:
+            return [text]
+
+        chunks = []
+        start = 0
+
+        while start < len(text):
+            end = start + chunk_size
+            chunk = text[start:end]
+            chunks.append(chunk)
+
+            # Move start position by chunk_size - overlap
+            start = end - overlap
+
+            # If remaining text is less than chunk_size, add it as final chunk
+            if len(text) - start < chunk_size:
+                if start < len(text):
+                    final_chunk = text[start:]
+                    if final_chunk not in chunks:  # Avoid duplicate chunks
+                        chunks.append(final_chunk)
+                break
+
+        return chunks
+
+    def embed(self, text: str) -> List[float]:
+        """
+        Generate embedding for text using Cohere
+        """
+        try:
+            response = self.cohere_client.embed(
+                texts=[text],
+                model="embed-multilingual-v3.0",  # Using multilingual model
+                input_type="search_document"  # Optimize for search
+            )
+            return response.embeddings[0]  # Return the first (and only) embedding
+        except Exception as e:
+            logger.error(f"Error generating embedding for text: {e}")
+            return []
+
+    def create_collection(self, collection_name: str = "rag_embedding"):
+        """
+        Create a Qdrant collection for storing embeddings
+        """
+        try:
+            # Check if collection already exists
+            collections = self.qdrant_client.get_collections()
+            collection_names = [col.name for col in collections.collections]
+
+            if collection_name in collection_names:
+                logger.info(f"Collection {collection_name} already exists")
+                return
+
+            # Create collection with appropriate vector size (1024 for Cohere embeddings)
+            self.qdrant_client.create_collection(
+                collection_name=collection_name,
+                vectors_config=models.VectorParams(size=1024, distance=models.Distance.COSINE)
+            )
+
+            logger.info(f"Created collection {collection_name} with 1024-dimension vectors")
+
+        except Exception as e:
+            logger.error(f"Error creating collection {collection_name}: {e}")
+            raise
+
+    def save_chunk_to_qdrant(self, content: str, url: str, embedding: List[float], position: int, collection_name: str = "rag_embedding"):
+        """
+        Save a text chunk with its embedding to Qdrant
+        """
+        try:
+            # Generate a unique ID for the point
+            point_id = str(uuid.uuid4())
+
+            # Prepare the payload with metadata
+            payload = {
+                "content": content,
+                "url": url,
+                "position": position,
+                "created_at": time.time()
+            }
+
+            # Create and upload the point to Qdrant
+            points = [PointStruct(
+                id=point_id,
+                vector=embedding,
+                payload=payload
+            )]
+
+            self.qdrant_client.upsert(
+                collection_name=collection_name,
+                points=points
+            )
+
+            logger.info(f"Saved chunk to Qdrant: {url} (position {position})")
+            return True
+
+        except Exception as e:
+            logger.error(f"Error saving chunk to Qdrant: {e}")
+            return False
+
+def main():
+    """
+    Main function to execute the complete pipeline
+    """
+    logger.info("Starting Docusaurus Embedding Pipeline...")
+
+    # Initialize the pipeline
+    pipeline = DocusaurusEmbeddingPipeline()
+
+    try:
+        # Step 1: Create the Qdrant collection
+        logger.info("Creating Qdrant collection...")
+        pipeline.create_collection("rag_embedding")
+
+        # Step 2: Get all URLs from the target Docusaurus site
+        logger.info(f"Extracting URLs from {pipeline.target_url}...")
+        urls = pipeline.get_all_urls(pipeline.target_url)
+
+        if not urls:
+            logger.warning(f"No URLs found at {pipeline.target_url}")
+            return
+
+        logger.info(f"Found {len(urls)} URLs to process")
+
+        # Step 3: Process each URL
+        total_chunks = 0
+        for i, url in enumerate(urls):
+            logger.info(f"Processing URL {i+1}/{len(urls)}: {url}")
+
+            # Extract text from the URL
+            text_content = pipeline.extract_text_from_url(url)
+
+            if not text_content:
+                logger.warning(f"No content extracted from {url}")
+                continue
+
+            logger.info(f"Extracted {len(text_content)} characters from {url}")
+
+            # Chunk the text
+            chunks = pipeline.chunk_text(text_content)
+            logger.info(f"Created {len(chunks)} chunks from {url}")
+
+            # Process each chunk
+            for j, chunk in enumerate(chunks):
+                if not chunk.strip():
+                    continue
+
+                # Generate embedding
+                embedding = pipeline.embed(chunk)
+
+                if not embedding:
+                    logger.error(f"Failed to generate embedding for chunk {j} of {url}")
+                    continue
+
+                # Save to Qdrant
+                success = pipeline.save_chunk_to_qdrant(
+                    content=chunk,
+                    url=url,
+                    embedding=embedding,
+                    position=j
+                )
+
+                if success:
+                    total_chunks += 1
+                    logger.info(f"Successfully saved chunk {j} of {url} to Qdrant")
+                else:
+                    logger.error(f"Failed to save chunk {j} of {url} to Qdrant")
+
+        logger.info(f"Pipeline completed successfully! Total chunks saved: {total_chunks}")
+
+    except Exception as e:
+        logger.error(f"Pipeline failed with error: {e}")
+        raise
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -0,0 +1,27 @@
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools"
+
+[project]
+name = "docusaurus-embedding-pipeline"
+version = "0.1.0"
+description = "Pipeline to extract text from Docusaurus sites, generate embeddings, and store in Qdrant"
+readme = "README.md"
+requires-python = ">=3.9"
+dependencies = [
+    "requests>=2.31.0",
+    "beautifulsoup4>=4.12.2",
+    "cohere>=4.9.0",
+    "qdrant-client>=1.9.0",
+    "python-dotenv>=1.0.0"
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=7.0",
+    "black>=23.0",
+    "flake8>=6.0"
+]
+
+[tool.setuptools.packages.find]
+where = ["."]

requirements.txt

@@ -0,0 +1,22 @@
+fastapi==0.104.1
+uvicorn[standard]==0.24.0
+python-dotenv==1.0.0
+qdrant-client==1.9.1
+httpx==0.25.2
+psycopg2-binary==2.9.9
+sqlalchemy==2.0.23
+pydantic==2.5.0
+pydantic-settings==2.1.0
+openai==1.3.6
+tiktoken==0.5.2
+markdown==3.5.1
+python-multipart==0.0.6
+python-jose[cryptography]==3.3.0
+passlib[bcrypt]==1.7.4
+python-slugify==8.0.1
+asyncpg==0.29.0
+alembic==1.13.1
+beautifulsoup4==4.12.2
+scikit-learn==1.3.2
+requests>=2.31.0
+cohere>=4.9.0

retrieving.py

@@ -0,0 +1,267 @@
+import os
+import json
+from typing import List, Dict, Any
+import cohere
+from qdrant_client import QdrantClient
+from qdrant_client.http import models
+import logging
+from dotenv import load_dotenv
+import time
+
+# Load environment variables
+load_dotenv()
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class RAGRetriever:
+    def __init__(self):
+        # Initialize Cohere client
+        self.cohere_client = cohere.Client(api_key=os.getenv("COHERE_API_KEY"))
+
+        # Initialize Qdrant client
+        qdrant_url = os.getenv("QDRANT_URL", "http://localhost:6333")
+        qdrant_api_key = os.getenv("QDRANT_API_KEY")
+
+        if qdrant_api_key:
+            self.qdrant_client = QdrantClient(url=qdrant_url, api_key=qdrant_api_key)
+        else:
+            self.qdrant_client = QdrantClient(url=qdrant_url)
+
+        # Default collection name
+        self.collection_name = "rag_embedding"
+
+    def get_embedding(self, text: str) -> List[float]:
+        """
+        Generate embedding for query text using Cohere
+        """
+        try:
+            response = self.cohere_client.embed(
+                texts=[text],
+                model="embed-multilingual-v3.0",  # Using same model as storage
+                input_type="search_query"  # Optimize for search queries
+            )
+            return response.embeddings[0]  # Return the first (and only) embedding
+        except Exception as e:
+            logger.error(f"Error generating embedding for query: {e}")
+            return []
+
+    def query_qdrant(self, query_embedding: List[float], top_k: int = 5, threshold: float = 0.0) -> List[Dict]:
+        """
+        Query Qdrant for similar vectors and return results with metadata
+        """
+        try:
+            # Perform similarity search in Qdrant
+            search_results = self.qdrant_client.search(
+                collection_name=self.collection_name,
+                query_vector=query_embedding,
+                limit=top_k,
+                score_threshold=threshold,
+                with_payload=True  # Include metadata with results
+            )
+
+            # Format results
+            formatted_results = []
+            for result in search_results:
+                formatted_result = {
+                    "content": result.payload.get("content", ""),
+                    "url": result.payload.get("url", ""),
+                    "position": result.payload.get("position", 0),
+                    "similarity_score": result.score,
+                    "chunk_id": result.id,
+                    "created_at": result.payload.get("created_at", "")
+                }
+                formatted_results.append(formatted_result)
+
+            return formatted_results
+
+        except Exception as e:
+            logger.error(f"Error querying Qdrant: {e}")
+            return []
+
+    def verify_content_accuracy(self, retrieved_chunks: List[Dict]) -> bool:
+        """
+        Verify that retrieved content matches original stored text (basic validation)
+        """
+        # In a real implementation, this would compare against original sources
+        # For now, we'll validate that required fields exist and have content
+        for chunk in retrieved_chunks:
+            if not chunk.get("content") or not chunk.get("url"):
+                logger.warning(f"Missing content or URL in chunk: {chunk.get('chunk_id')}")
+                return False
+
+        # Additional validation could include checking content length, URL format, etc.
+        return True
+
+    def format_json_response(self, results: List[Dict], query: str, query_time_ms: float) -> str:
+        """
+        Format retrieval results into clean JSON response
+        """
+        response = {
+            "query": query,
+            "results": results,
+            "metadata": {
+                "query_time_ms": query_time_ms,
+                "total_results": len(results),
+                "timestamp": time.time(),
+                "collection_name": self.collection_name
+            }
+        }
+
+        return json.dumps(response, indent=2)
+
+    def retrieve(self, query_text: str, top_k: int = 5, threshold: float = 0.0, include_metadata: bool = True) -> str:
+        """
+        Main retrieval function that orchestrates the complete workflow
+        """
+        start_time = time.time()
+
+        logger.info(f"Processing retrieval request for query: '{query_text[:50]}...'")
+
+        # Step 1: Convert query text to embedding
+        query_embedding = self.get_embedding(query_text)
+        if not query_embedding:
+            error_response = {
+                "query": query_text,
+                "results": [],
+                "error": "Failed to generate query embedding",
+                "metadata": {
+                    "query_time_ms": (time.time() - start_time) * 1000,
+                    "timestamp": time.time()
+                }
+            }
+            return json.dumps(error_response, indent=2)
+
+        # Step 2: Query Qdrant for similar vectors
+        raw_results = self.query_qdrant(query_embedding, top_k, threshold)
+
+        if not raw_results:
+            logger.warning("No results returned from Qdrant")
+
+        # Step 3: Verify content accuracy (optional)
+        if include_metadata:
+            is_accurate = self.verify_content_accuracy(raw_results)
+            if not is_accurate:
+                logger.warning("Content accuracy verification failed for some results")
+
+        # Step 4: Calculate total query time
+        query_time_ms = (time.time() - start_time) * 1000
+
+        # Step 5: Format response as JSON
+        json_response = self.format_json_response(raw_results, query_text, query_time_ms)
+
+        logger.info(f"Retrieval completed in {query_time_ms:.2f}ms, {len(raw_results)} results returned")
+
+        return json_response
+
+def retrieve_all_data():
+    """
+    Function to retrieve and display all data from Qdrant collection
+    """
+    logger.info("Initializing RAG Retriever to fetch all data...")
+
+    # Initialize the retriever
+    retriever = RAGRetriever()
+
+    print("RAG Retrieval System - All Stored Data")
+    print("=" * 50)
+
+    try:
+        # Get all points from the collection using scroll
+        points = []
+        offset = None
+        while True:
+            # Scroll through the collection to get all points
+            batch, next_offset = retriever.qdrant_client.scroll(
+                collection_name=retriever.collection_name,
+                limit=1000,  # Get up to 1000 points at a time
+                offset=offset,
+                with_payload=True,
+                with_vectors=False
+            )
+
+            points.extend(batch)
+
+            # If next_offset is None, we've reached the end
+            if next_offset is None:
+                break
+
+            offset = next_offset
+
+        print(f"Total stored chunks: {len(points)}")
+        print("-" * 50)
+
+        for i, point in enumerate(points, 1):
+            payload = point.payload
+            content_preview = ''.join(char for char in payload.get("content", "")[:200] if ord(char) < 256)
+
+            print(f"Chunk {i}:")
+            print(f"  ID: {point.id}")
+            print(f"  URL: {payload.get('url', 'N/A')}")
+            print(f"  Position: {payload.get('position', 'N/A')}")
+            print(f"  Content Preview: {content_preview}...")
+            print(f"  Created At: {payload.get('created_at', 'N/A')}")
+            print("-" * 30)
+
+    except Exception as e:
+        logger.error(f"Error retrieving all data: {e}")
+        print(f"Error retrieving all data: {e}")
+
+
+def main():
+    """
+    Main function to demonstrate the retrieval functionality
+    """
+    import sys
+
+    logger.info("Initializing RAG Retriever...")
+
+    # Check if user wants to retrieve all data or run queries
+    if len(sys.argv) > 1 and sys.argv[1] == "all":
+        retrieve_all_data()
+        return
+
+    # Initialize the retriever
+    retriever = RAGRetriever()
+
+    # Example queries to test the system
+    test_queries = [
+        "What is ROS2?",
+        "Explain humanoid design principles",
+        "How does VLA work?",
+        "What are simulation techniques?",
+        "Explain AI control systems"
+    ]
+
+    print("RAG Retrieval System - Testing Queries")
+    print("=" * 50)
+
+    for i, query in enumerate(test_queries, 1):
+        print(f"\nQuery {i}: {query}")
+        print("-" * 30)
+
+        # Retrieve results
+        json_response = retriever.retrieve(query, top_k=3)
+        response_dict = json.loads(json_response)
+
+        # Print formatted results
+        results = response_dict.get("results", [])
+        if results:
+            for j, result in enumerate(results, 1):
+                print(f"Result {j} (Score: {result['similarity_score']:.3f}):")
+                print(f"  URL: {result['url']}")
+                content_preview = result['content'][:100].encode('utf-8', errors='ignore').decode('utf-8')
+                # Safely print content preview by removing problematic characters
+                safe_content = ''.join(char for char in content_preview if ord(char) < 256)
+                print(f"  Content Preview: {safe_content}...")
+                print(f"  Position: {result['position']}")
+                print()
+        else:
+            print("No results found for this query.")
+
+        print(f"Query time: {response_dict['metadata']['query_time_ms']:.2f}ms")
+        print(f"Total results: {response_dict['metadata']['total_results']}")
+
+if __name__ == "__main__":
+    main()

sdk.md

@@ -0,0 +1,935 @@
+# OpenAI Agents SDK
+
+The OpenAI Agents SDK is a lightweight Python framework for building production-ready agentic AI applications with minimal abstractions. It provides a streamlined upgrade from the experimental Swarm framework, offering essential primitives like agents with instructions and tools, handoffs for task delegation between agents, guardrails for input/output validation, and sessions for automatic conversation history management. The SDK emphasizes ease of use while maintaining enough power to express complex multi-agent relationships, making it suitable for real-world applications without requiring mastery of complex frameworks.
+
+Built on core design principles of simplicity and customization, the SDK includes an automatic agent loop handling tool calls and LLM interactions, Python-first orchestration without new abstractions to learn, built-in tracing for visualization and debugging, and automatic schema generation with Pydantic validation for function tools. It supports multiple model providers through OpenAI's Responses API and Chat Completions API, with native integration for LiteLLM and custom providers. Whether building single-agent assistants or complex multi-agent workflows with specialized roles, the SDK provides the necessary features to move quickly from prototype to production.
+
+## Installation and Setup
+
+Install the SDK and configure your environment
+
+```bash
+pip install openai-agents
+
+export OPENAI_API_KEY=sk-...
+```
+
+## Creating a Basic Agent
+
+Define an agent with name and instructions
+
+```python
+from agents import Agent, Runner
+
+agent = Agent(
+    name="Math Tutor",
+    instructions="You provide help with math problems. Explain your reasoning at each step and include examples"
+)
+
+result = Runner.run_sync(agent, "What is 15% of 80?")
+print(result.final_output)
+# 15% of 80 is 12. To calculate: 0.15 × 80 = 12
+```
+
+## Running Agents Asynchronously
+
+Execute agent with async/await pattern
+
+```python
+import asyncio
+from agents import Agent, Runner
+
+async def main():
+    agent = Agent(
+        name="Assistant",
+        instructions="Reply very concisely."
+    )
+
+    result = await Runner.run(agent, "What city is the Golden Gate Bridge in?")
+    print(result.final_output)
+    # San Francisco
+
+asyncio.run(main())
+```
+
+## Agent with Function Tools
+
+Decorate Python functions to create tools with automatic schema generation
+
+```python
+from agents import Agent, Runner, function_tool
+import asyncio
+
+@function_tool
+async def get_weather(city: str) -> str:
+    """Fetch the weather for a given location.
+
+    Args:
+        city: The city to fetch weather for.
+    """
+    # In production, call actual weather API
+    return f"The weather in {city} is sunny and 72°F"
+
+@function_tool
+def calculate_sum(a: int, b: int) -> int:
+    """Add two numbers together.
+
+    Args:
+        a: First number
+        b: Second number
+    """
+    return a + b
+
+agent = Agent(
+    name="Assistant",
+    instructions="Use the provided tools to help the user",
+    tools=[get_weather, calculate_sum]
+)
+
+async def main():
+    result = await Runner.run(agent, "What's the weather in Seattle?")
+    print(result.final_output)
+    # The weather in Seattle is sunny and 72°F
+
+asyncio.run(main())
+```
+
+## Agent with Hosted Tools
+
+Use OpenAI's built-in tools for web search and file retrieval
+
+```python
+from agents import Agent, Runner, WebSearchTool, FileSearchTool
+import asyncio
+
+agent = Agent(
+    name="Research Assistant",
+    instructions="Use web search and file search to answer questions thoroughly",
+    tools=[
+        WebSearchTool(),
+        FileSearchTool(
+            max_num_results=5,
+            vector_store_ids=["vs_abc123"]
+        )
+    ]
+)
+
+async def main():
+    result = await Runner.run(
+        agent,
+        "What are the latest developments in quantum computing?"
+    )
+    print(result.final_output)
+
+asyncio.run(main())
+```
+
+## Multi-Agent Handoffs
+
+Create specialized agents that delegate to each other
+
+```python
+from agents import Agent, Runner
+import asyncio
+
+billing_agent = Agent(
+    name="Billing Agent",
+    handoff_description="Specialist for billing questions and payment issues",
+    instructions="You handle billing inquiries. Check account status and process refunds."
+)
+
+technical_agent = Agent(
+    name="Technical Agent",
+    handoff_description="Specialist for technical support and troubleshooting",
+    instructions="You handle technical issues. Diagnose problems and provide solutions."
+)
+
+triage_agent = Agent(
+    name="Triage Agent",
+    instructions=(
+        "Determine which specialist agent should handle the user's request. "
+        "Hand off to the appropriate agent based on the question type."
+    ),
+    handoffs=[billing_agent, technical_agent]
+)
+
+async def main():
+    result = await Runner.run(
+        triage_agent,
+        "I was charged twice for my subscription this month"
+    )
+    print(result.final_output)
+    # Output from billing_agent after handoff
+
+asyncio.run(main())
+```
+
+## Custom Handoff with Input Data
+
+Configure handoffs with structured input and callbacks
+
+```python
+from agents import Agent, Runner, handoff, RunContextWrapper
+from pydantic import BaseModel
+import asyncio
+
+class EscalationData(BaseModel):
+    reason: str
+    severity: str
+
+async def on_escalation(ctx: RunContextWrapper[None], input_data: EscalationData):
+    print(f"Escalated: {input_data.reason} (severity: {input_data.severity})")
+    # Log to monitoring system, send alert, etc.
+
+escalation_agent = Agent(
+    name="Manager",
+    instructions="Handle escalated customer issues with priority"
+)
+
+support_agent = Agent(
+    name="Support Agent",
+    instructions="Help customers. Escalate to manager if issue is severe.",
+    handoffs=[
+        handoff(
+            agent=escalation_agent,
+            on_handoff=on_escalation,
+            input_type=EscalationData,
+            tool_description_override="Escalate urgent issues to management"
+        )
+    ]
+)
+
+async def main():
+    result = await Runner.run(
+        support_agent,
+        "This is completely unacceptable! I demand to speak to a manager!"
+    )
+    print(result.final_output)
+
+asyncio.run(main())
+```
+
+## Input Guardrails
+
+Validate user input before processing with the main agent
+
+```python
+from agents import Agent, Runner, input_guardrail, GuardrailFunctionOutput
+from agents import InputGuardrailTripwireTriggered, RunContextWrapper, TResponseInputItem
+from pydantic import BaseModel
+import asyncio
+
+class HomeworkCheck(BaseModel):
+    is_homework: bool
+    reasoning: str
+
+guardrail_agent = Agent(
+    name="Homework Detector",
+    instructions="Determine if the user is asking for homework help",
+    output_type=HomeworkCheck
+)
+
+@input_guardrail
+async def homework_guardrail(
+    ctx: RunContextWrapper[None],
+    agent: Agent,
+    input_data: str | list[TResponseInputItem]
+) -> GuardrailFunctionOutput:
+    result = await Runner.run(guardrail_agent, input_data, context=ctx.context)
+
+    return GuardrailFunctionOutput(
+        output_info=result.final_output,
+        tripwire_triggered=result.final_output.is_homework
+    )
+
+tutoring_agent = Agent(
+    name="Tutoring Service",
+    instructions="You help students understand concepts, not do their homework",
+    input_guardrails=[homework_guardrail]
+)
+
+async def main():
+    try:
+        result = await Runner.run(
+            tutoring_agent,
+            "Can you solve this equation for me: 2x + 5 = 15?"
+        )
+        print(result.final_output)
+    except InputGuardrailTripwireTriggered as e:
+        print("Request blocked: This appears to be homework help")
+
+asyncio.run(main())
+```
+
+## Output Guardrails
+
+Validate agent responses before returning to user
+
+```python
+from agents import Agent, Runner, output_guardrail, GuardrailFunctionOutput
+from agents import OutputGuardrailTripwireTriggered, RunContextWrapper
+from pydantic import BaseModel
+import asyncio
+
+class ToxicityCheck(BaseModel):
+    is_toxic: bool
+    confidence: float
+
+class AgentResponse(BaseModel):
+    message: str
+
+toxicity_checker = Agent(
+    name="Toxicity Detector",
+    instructions="Analyze if the message contains toxic or harmful content",
+    output_type=ToxicityCheck
+)
+
+@output_guardrail
+async def toxicity_guardrail(
+    ctx: RunContextWrapper[None],
+    agent: Agent,
+    output: AgentResponse
+) -> GuardrailFunctionOutput:
+    result = await Runner.run(toxicity_checker, output.message, context=ctx.context)
+
+    return GuardrailFunctionOutput(
+        output_info=result.final_output,
+        tripwire_triggered=result.final_output.is_toxic and result.final_output.confidence > 0.8
+    )
+
+chatbot = Agent(
+    name="Chatbot",
+    instructions="You are a friendly assistant",
+    output_guardrails=[toxicity_guardrail],
+    output_type=AgentResponse
+)
+
+async def main():
+    try:
+        result = await Runner.run(chatbot, "Tell me about your day")
+        print(result.final_output.message)
+    except OutputGuardrailTripwireTriggered:
+        print("Response blocked by content filter")
+
+asyncio.run(main())
+```
+
+## Sessions for Conversation Memory
+
+Automatically maintain conversation history across multiple turns
+
+```python
+from agents import Agent, Runner, SQLiteSession
+import asyncio
+
+async def main():
+    agent = Agent(
+        name="Assistant",
+        instructions="Reply concisely and remember previous context"
+    )
+
+    # Create persistent session with SQLite backend
+    session = SQLiteSession("user_123", "conversations.db")
+
+    # First turn
+    result = await Runner.run(
+        agent,
+        "What city is the Golden Gate Bridge in?",
+        session=session
+    )
+    print(result.final_output)
+    # San Francisco
+
+    # Second turn - agent remembers previous context
+    result = await Runner.run(
+        agent,
+        "What state is it in?",
+        session=session
+    )
+    print(result.final_output)
+    # California
+
+    # Third turn - continuing the conversation
+    result = await Runner.run(
+        agent,
+        "What's the population?",
+        session=session
+    )
+    print(result.final_output)
+    # Approximately 39 million
+
+asyncio.run(main())
+```
+
+## Session Management Operations
+
+Manipulate conversation history programmatically
+
+```python
+from agents import Agent, Runner, SQLiteSession
+import asyncio
+
+async def main():
+    session = SQLiteSession("conversation_456", "chats.db")
+
+    # Get all conversation items
+    items = await session.get_items()
+    print(f"Total messages: {len(items)}")
+
+    # Add items manually
+    await session.add_items([
+        {"role": "user", "content": "Hello"},
+        {"role": "assistant", "content": "Hi! How can I help?"}
+    ])
+
+    # Remove last item (useful for corrections)
+    agent = Agent(name="Assistant")
+
+    result = await Runner.run(agent, "What's 2 + 2?", session=session)
+    print(result.final_output)
+
+    # User wants to correct their question
+    await session.pop_item()  # Remove assistant response
+    await session.pop_item()  # Remove user question
+
+    result = await Runner.run(agent, "What's 2 + 3?", session=session)
+    print(result.final_output)
+
+    # Clear entire session
+    await session.clear_session()
+
+asyncio.run(main())
+```
+
+## OpenAI Conversations Session
+
+Use OpenAI-hosted conversation storage
+
+```python
+from agents import Agent, Runner, OpenAIConversationsSession
+import asyncio
+
+async def main():
+    agent = Agent(name="Assistant")
+
+    # Create new conversation or resume existing one
+    session = OpenAIConversationsSession()
+    # Or with existing conversation ID:
+    # session = OpenAIConversationsSession(conversation_id="conv_abc123")
+
+    result = await Runner.run(
+        agent,
+        "Remember that my favorite color is blue",
+        session=session
+    )
+
+    # Later conversation with same session
+    result = await Runner.run(
+        agent,
+        "What's my favorite color?",
+        session=session
+    )
+    print(result.final_output)
+    # Your favorite color is blue
+
+asyncio.run(main())
+```
+
+## Structured Outputs
+
+Force agents to return specific data types with validation
+
+```python
+from agents import Agent, Runner
+from pydantic import BaseModel
+import asyncio
+
+class CalendarEvent(BaseModel):
+    title: str
+    date: str
+    participants: list[str]
+    location: str | None = None
+
+agent = Agent(
+    name="Calendar Parser",
+    instructions="Extract calendar event information from text",
+    output_type=CalendarEvent
+)
+
+async def main():
+    text = "Schedule a team meeting on March 15th with John, Sarah, and Mike"
+    result = await Runner.run(agent, text)
+
+    event = result.final_output_as(CalendarEvent)
+    print(f"Event: {event.title}")
+    print(f"Date: {event.date}")
+    print(f"Attendees: {', '.join(event.participants)}")
+    # Event: Team Meeting
+    # Date: March 15th
+    # Attendees: John, Sarah, Mike
+
+asyncio.run(main())
+```
+
+## Agent Context and Dependency Injection
+
+Pass custom context objects to agents and tools
+
+```python
+from dataclasses import dataclass
+from agents import Agent, Runner, RunContextWrapper, function_tool
+import asyncio
+
+@dataclass
+class UserContext:
+    user_id: str
+    is_premium: bool
+    api_token: str
+
+@function_tool
+async def get_user_data(ctx: RunContextWrapper[UserContext]) -> str:
+    """Fetch user-specific data using context."""
+    user_id = ctx.context.user_id
+    is_premium = ctx.context.is_premium
+
+    if is_premium:
+        return f"Premium user {user_id} has access to all features"
+    return f"User {user_id} has basic access"
+
+agent = Agent[UserContext](
+    name="Account Manager",
+    instructions="Provide user information based on their account status",
+    tools=[get_user_data]
+)
+
+async def main():
+    context = UserContext(
+        user_id="user_789",
+        is_premium=True,
+        api_token="secret_token"
+    )
+
+    result = await Runner.run(agent, "What's my account status?", context=context)
+    print(result.final_output)
+
+asyncio.run(main())
+```
+
+## Dynamic Instructions
+
+Generate agent instructions at runtime based on context
+
+```python
+from agents import Agent, Runner, RunContextWrapper
+from dataclasses import dataclass
+import asyncio
+
+@dataclass
+class AppContext:
+    username: str
+    language: str
+    timezone: str
+
+def dynamic_instructions(
+    context: RunContextWrapper[AppContext],
+    agent: Agent[AppContext]
+) -> str:
+    user = context.context
+    return f"""You are a helpful assistant for {user.username}.
+    - Respond in {user.language}
+    - Use {user.timezone} timezone for all time references
+    - Be friendly and personalized"""
+
+agent = Agent[AppContext](
+    name="Personal Assistant",
+    instructions=dynamic_instructions
+)
+
+async def main():
+    context = AppContext(
+        username="Alice",
+        language="Spanish",
+        timezone="PST"
+    )
+
+    result = await Runner.run(agent, "What time is it?", context=context)
+    print(result.final_output)
+
+asyncio.run(main())
+```
+
+## Streaming Agent Responses
+
+Stream token-by-token responses from the agent
+
+```python
+from agents import Agent, Runner
+from openai.types.responses import ResponseTextDeltaEvent
+import asyncio
+
+async def main():
+    agent = Agent(
+        name="Storyteller",
+        instructions="Tell engaging short stories"
+    )
+
+    result = Runner.run_streamed(agent, "Tell me a story about a robot")
+
+    print("Streaming response: ", end="", flush=True)
+    async for event in result.stream_events():
+        if event.type == "raw_response_event":
+            if isinstance(event.data, ResponseTextDeltaEvent):
+                print(event.data.delta, end="", flush=True)
+
+    print("\n\nFinal output:", result.final_output)
+
+asyncio.run(main())
+```
+
+## Streaming with Item-Level Events
+
+Stream higher-level events like tool calls and messages
+
+```python
+from agents import Agent, Runner, ItemHelpers, function_tool
+import asyncio
+import random
+
+@function_tool
+def roll_dice(sides: int = 6) -> int:
+    """Roll a dice with specified number of sides."""
+    return random.randint(1, sides)
+
+async def main():
+    agent = Agent(
+        name="Game Master",
+        instructions="Use the dice rolling tool when asked",
+        tools=[roll_dice]
+    )
+
+    result = Runner.run_streamed(agent, "Roll two dice for me")
+
+    async for event in result.stream_events():
+        if event.type == "raw_response_event":
+            continue  # Skip token-level events
+        elif event.type == "agent_updated_stream_event":
+            print(f"Agent: {event.new_agent.name}")
+        elif event.type == "run_item_stream_event":
+            if event.item.type == "tool_call_item":
+                print("🔧 Tool called")
+            elif event.item.type == "tool_call_output_item":
+                print(f"📤 Tool result: {event.item.output}")
+            elif event.item.type == "message_output_item":
+                text = ItemHelpers.text_message_output(event.item)
+                print(f"💬 Agent: {text}")
+
+asyncio.run(main())
+```
+
+## Agents as Tools Pattern
+
+Use specialized agents as tools in a central orchestrator
+
+```python
+from agents import Agent, Runner
+import asyncio
+
+translation_agent = Agent(
+    name="Translator",
+    instructions="Translate the user's message to the specified language"
+)
+
+summarization_agent = Agent(
+    name="Summarizer",
+    instructions="Create a concise summary of the provided text"
+)
+
+orchestrator = Agent(
+    name="Orchestrator",
+    instructions="Use the available tools to process user requests efficiently",
+    tools=[
+        translation_agent.as_tool(
+            tool_name="translate_text",
+            tool_description="Translate text to another language"
+        ),
+        summarization_agent.as_tool(
+            tool_name="summarize_text",
+            tool_description="Generate a summary of long text"
+        )
+    ]
+)
+
+async def main():
+    result = await Runner.run(
+        orchestrator,
+        "Translate 'Hello, how are you?' to French and Spanish"
+    )
+    print(result.final_output)
+
+asyncio.run(main())
+```
+
+## Custom Model Configuration
+
+Configure model settings and use different models per agent
+
+```python
+from agents import Agent, Runner, ModelSettings
+from openai.types.shared import Reasoning
+import asyncio
+
+reasoning_agent = Agent(
+    name="Deep Thinker",
+    instructions="Analyze complex problems thoroughly",
+    model="gpt-5",
+    model_settings=ModelSettings(
+        reasoning=Reasoning(effort="high"),
+        temperature=0.7,
+        verbosity="high"
+    )
+)
+
+fast_agent = Agent(
+    name="Quick Responder",
+    instructions="Provide rapid responses",
+    model="gpt-5-nano",
+    model_settings=ModelSettings(
+        reasoning=Reasoning(effort="minimal"),
+        temperature=0.3,
+        verbosity="low"
+    )
+)
+
+triage_agent = Agent(
+    name="Router",
+    instructions="Route complex problems to deep thinker, simple ones to quick responder",
+    handoffs=[reasoning_agent, fast_agent]
+)
+
+async def main():
+    result = await Runner.run(
+        triage_agent,
+        "Explain quantum entanglement in simple terms"
+    )
+    print(result.final_output)
+
+asyncio.run(main())
+```
+
+## MCP Hosted Tool Integration
+
+Use Model Context Protocol servers as hosted tools
+
+```python
+from agents import Agent, Runner, HostedMCPTool
+import asyncio
+
+async def main():
+    agent = Agent(
+        name="Code Assistant",
+        instructions="Help with repository questions using git tools",
+        tools=[
+            HostedMCPTool(
+                tool_config={
+                    "type": "mcp",
+                    "server_label": "gitmcp",
+                    "server_url": "https://gitmcp.io/openai/codex",
+                    "require_approval": "never"
+                }
+            )
+        ]
+    )
+
+    result = await Runner.run(
+        agent,
+        "What programming languages are used in this repository?"
+    )
+    print(result.final_output)
+
+asyncio.run(main())
+```
+
+## MCP Server with Streamable HTTP
+
+Connect to local or remote MCP servers via HTTP
+
+```python
+from agents import Agent, Runner, ModelSettings
+from agents.mcp import MCPServerStreamableHttp
+import asyncio
+import os
+
+async def main():
+    token = os.environ["MCP_SERVER_TOKEN"]
+
+    async with MCPServerStreamableHttp(
+        name="Calculator Server",
+        params={
+            "url": "http://localhost:8000/mcp",
+            "headers": {"Authorization": f"Bearer {token}"},
+            "timeout": 10
+        },
+        cache_tools_list=True,
+        max_retry_attempts=3
+    ) as server:
+        agent = Agent(
+            name="Math Assistant",
+            instructions="Use MCP tools to perform calculations",
+            mcp_servers=[server],
+            model_settings=ModelSettings(tool_choice="required")
+        )
+
+        result = await Runner.run(agent, "Calculate 47 + 89")
+        print(result.final_output)
+
+asyncio.run(main())
+```
+
+## MCP stdio Server
+
+Launch local MCP server processes
+
+```python
+from agents import Agent, Runner
+from agents.mcp import MCPServerStdio
+from pathlib import Path
+import asyncio
+
+async def main():
+    samples_dir = Path(__file__).parent / "sample_files"
+
+    async with MCPServerStdio(
+        name="Filesystem Server",
+        params={
+            "command": "npx",
+            "args": ["-y", "@modelcontextprotocol/server-filesystem", str(samples_dir)]
+        }
+    ) as server:
+        agent = Agent(
+            name="File Assistant",
+            instructions="Help users work with files in the sample directory",
+            mcp_servers=[server]
+        )
+
+        result = await Runner.run(agent, "List all files in the directory")
+        print(result.final_output)
+
+asyncio.run(main())
+```
+
+## Tracing and Monitoring
+
+Built-in tracing for debugging and monitoring agent workflows
+
+```python
+from agents import Agent, Runner, trace
+import asyncio
+
+async def main():
+    agent = Agent(
+        name="Research Agent",
+        instructions="Research topics thoroughly"
+    )
+
+    # Trace multiple runs under single workflow
+    with trace(
+        workflow_name="Research Workflow",
+        group_id="session_123",
+        metadata={"user": "alice", "environment": "production"}
+    ):
+        result1 = await Runner.run(agent, "What is machine learning?")
+        print(f"Response 1: {result1.final_output}")
+
+        result2 = await Runner.run(agent, "Explain neural networks")
+        print(f"Response 2: {result2.final_output}")
+
+    # View traces at: https://platform.openai.com/traces
+
+asyncio.run(main())
+```
+
+## Error Handling
+
+Handle exceptions from agent runs, guardrails, and tool failures
+
+```python
+from agents import Agent, Runner, function_tool
+from agents.exceptions import (
+    MaxTurnsExceeded,
+    InputGuardrailTripwireTriggered,
+    ModelBehaviorError
+)
+import asyncio
+
+@function_tool
+def risky_operation() -> str:
+    """An operation that might fail."""
+    raise ValueError("Operation failed!")
+
+agent = Agent(
+    name="Assistant",
+    instructions="Help users with tasks",
+    tools=[risky_operation]
+)
+
+async def main():
+    try:
+        result = await Runner.run(
+            agent,
+            "Run the risky operation",
+            max_turns=5
+        )
+        print(result.final_output)
+
+    except MaxTurnsExceeded:
+        print("Error: Agent exceeded maximum turns")
+    except InputGuardrailTripwireTriggered as e:
+        print(f"Error: Input blocked by guardrail: {e}")
+    except ModelBehaviorError as e:
+        print(f"Error: Model produced invalid output: {e}")
+    except Exception as e:
+        print(f"Unexpected error: {e}")
+
+asyncio.run(main())
+```
+
+## Using Alternative Model Providers
+
+Integrate non-OpenAI models via LiteLLM
+
+```bash
+pip install "openai-agents[litellm]"
+```
+
+```python
+from agents import Agent, Runner
+import asyncio
+
+async def main():
+    # Use Claude via LiteLLM
+    claude_agent = Agent(
+        name="Claude Assistant",
+        instructions="You are a helpful assistant",
+        model="litellm/anthropic/claude-3-5-sonnet-20240620"
+    )
+
+    # Use Gemini via LiteLLM
+    gemini_agent = Agent(
+        name="Gemini Assistant",
+        instructions="You are a helpful assistant",
+        model="litellm/gemini/gemini-2.5-flash-preview-04-17"
+    )
+
+    result = await Runner.run(claude_agent, "Explain photosynthesis briefly")
+    print(result.final_output)
+
+asyncio.run(main())
+```
+
+---
+
+## Summary
+
+The OpenAI Agents SDK provides a comprehensive yet simple framework for building agentic AI applications in Python. Core use cases include single-agent assistants with tool access, multi-agent systems with specialized roles using handoffs, conversational applications with automatic session memory, and workflows with input/output validation via guardrails. The SDK excels at building customer service bots with agent routing, research assistants with web search and file retrieval, code generation tools with MCP integration, and any application requiring LLM orchestration with minimal boilerplate.
+
+Integration patterns follow Python-first principles using native async/await, context managers for resource handling, decorators for function tools, and Pydantic models for structured outputs. The framework supports horizontal scaling through session persistence with SQLite or SQLAlchemy backends, vertical scaling with model mixing (fast models for triage, powerful models for complex tasks), and comprehensive observability through built-in tracing to OpenAI's dashboard or custom processors. Whether building prototypes or production systems, the SDK's balance of simplicity and power makes it an ideal choice for Python developers working with AI agents.