# Xaibo - The Modular AI Agent Framework > Build robust AI agents that are a joy to create and maintain. Stop fighting complexity, start shipping. Xaibo is a modular framework designed to help you build sophisticated AI systems that are easy to test, debug, and evolve. Move beyond monolithic agent scripts and start creating with components you can trust. # Tutorials # How-to Guides # How-to Guides How-to guides provide step-by-step instructions for solving specific problems with Xaibo. Each guide focuses on accomplishing a particular goal and assumes you have basic familiarity with Xaibo concepts. ## Installation and Setup - [How to install Xaibo with different dependency groups](installation/) - Install Xaibo with only the dependencies you need ## Security - [How to set up authentication](authentication/) - Secure your Xaibo server with API key authentication for OpenAI and MCP adapters ## Tools Integration - [How to create and integrate Python tools](tools/python-tools/) - Add custom Python functions as agent tools - [How to integrate MCP (Model Context Protocol) tools](tools/mcp-tools/) - Connect external MCP servers to your agents ## Integrations - [How to use Xaibo agents in LiveKit voice assistants](integrations/livekit-voice-assistant/) - Create real-time conversational AI applications with LiveKit ## LLM Configuration - [How to switch between different LLM providers](llm/switch-providers/) - Configure agents to use OpenAI, Anthropic, Google, or AWS Bedrock ## Memory and Storage - [How to set up vector memory for agents](memory/setup-vector-memory/) - Enable agents to store and retrieve information using vector embeddings ## Deployment - [How to deploy with OpenAI-compatible API](deployment/openai-api/) - Expose your agents through an OpenAI-compatible REST API - [How to deploy as an MCP server](deployment/mcp-server/) - Make your agents available as MCP tools for other applications - [How to start the OpenAI Responses API (Quickstart)](api/openai-responses-quickstart/) - Get the OpenAI Responses API running in 2 minutes ## Examples Explore complete working examples that demonstrate Xaibo in action: - [Google Calendar Example](https://github.com/xpressai/xaibo/tree/main/examples/google_calendar_example) - Build an agent that can read and create calendar events using Google Calendar API - [LiveKit Voice Assistant Example](https://github.com/xpressai/xaibo/tree/main/examples/livekit_example) - Create a real-time voice assistant using LiveKit for speech-to-text and text-to-speech ## Getting Help If you encounter issues while following these guides: 1. Check the [troubleshooting section](../reference/troubleshooting/) in the reference documentation 1. Review the [tutorial](../tutorial/) for foundational concepts 1. Join our [Discord community](https://discord.gg/uASMzSSVKe) for support 1. Report bugs on [GitHub](https://github.com/xpressai/xaibo/issues) # How to Set Up Authentication This guide shows you how to secure your Xaibo server with API key authentication for both OpenAI and MCP adapters. ## Quick Start The fastest way to add authentication: 1. **Set environment variables:** ```bash export CUSTOM_OPENAI_API_KEY="your-openai-secret-key" export MCP_API_KEY="your-mcp-secret-key" ``` 1. **Start the server:** ```bash python -m xaibo.server.web --adapter xaibo.server.adapters.OpenAiApiAdapter --adapter xaibo.server.adapters.McpApiAdapter ``` 1. **Test with curl:** ```bash curl -H "Authorization: Bearer your-openai-secret-key" http://localhost:8000/openai/models ``` ## Step-by-Step Setup ### 1. Choose Your Authentication Method **Option A: Environment Variables (Recommended)** ```bash export CUSTOM_OPENAI_API_KEY="sk-your-secret-key-here" export MCP_API_KEY="mcp-your-secret-key-here" ``` **Option B: Command Line Arguments** ```bash python -m xaibo.server.web \ --openai-api-key "sk-your-secret-key-here" \ --mcp-api-key "mcp-your-secret-key-here" \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --adapter xaibo.server.adapters.McpApiAdapter ``` ### 2. Start the Server with Adapters **For OpenAI API only:** ```bash python -m xaibo.server.web --adapter xaibo.server.adapters.OpenAiApiAdapter ``` **For MCP only:** ```bash python -m xaibo.server.web --adapter xaibo.server.adapters.McpApiAdapter ``` **For both adapters:** ```bash python -m xaibo.server.web \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --adapter xaibo.server.adapters.McpApiAdapter ``` ### 3. Configure Your Agent Directory ```bash python -m xaibo.server.web \ --agent-dir ./my-agents \ --adapter xaibo.server.adapters.OpenAiApiAdapter ``` ## Testing Authentication ### Test OpenAI Adapter **List available models:** ```bash curl -H "Authorization: Bearer your-openai-secret-key" \ http://localhost:8000/openai/models ``` **Send a chat completion:** ```bash curl -X POST \ -H "Authorization: Bearer your-openai-secret-key" \ -H "Content-Type: application/json" \ -d '{ "model": "your-agent-name", "messages": [{"role": "user", "content": "Hello"}] }' \ http://localhost:8000/openai/chat/completions ``` ### Test MCP Adapter **Initialize MCP connection:** ```bash curl -X POST \ -H "Authorization: Bearer your-mcp-secret-key" \ -H "Content-Type: application/json" \ -d '{ "jsonrpc": "2.0", "id": "1", "method": "initialize", "params": { "protocolVersion": "2024-11-05", "capabilities": {} } }' \ http://localhost:8000/mcp/ ``` **List available tools:** ```bash curl -X POST \ -H "Authorization: Bearer your-mcp-secret-key" \ -H "Content-Type: application/json" \ -d '{ "jsonrpc": "2.0", "id": "2", "method": "tools/list", "params": {} }' \ http://localhost:8000/mcp/ ``` ## Common Issues ### "Missing Authorization header" **Problem:** You forgot to include the Authorization header. **Solution:** Add the header to your request: ```bash curl -H "Authorization: Bearer your-api-key" http://localhost:8000/openai/models ``` ### "Invalid API key" **Problem:** The API key doesn't match what the server expects. **Solutions:** - Check your environment variables: `echo $CUSTOM_OPENAI_API_KEY` - Verify the key in your curl command matches exactly - Restart the server after changing environment variables ### Server starts but authentication doesn't work **Problem:** API key not configured properly. **Solutions:** - Make sure environment variables are set before starting the server - Use command-line arguments instead: `--openai-api-key "your-key"` - Check server logs for authentication errors ### MCP requests return HTTP 200 but with errors **Problem:** This is normal - MCP uses JSON-RPC protocol. **Solution:** Check the response body for error details: ```json { "jsonrpc": "2.0", "id": "1", "error": { "code": -32001, "message": "Invalid API key" } } ``` ## Security Notes - **Keep API keys secret** - never commit them to version control - **Use environment variables** for production deployments - **Use strong, random keys** - consider using `openssl rand -hex 32` - **Authentication is optional** - adapters work without API keys for development ## No Authentication Setup To run without authentication (development only): ```bash python -m xaibo.server.web \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --adapter xaibo.server.adapters.McpApiAdapter ``` Then test without Authorization headers: ```bash curl http://localhost:8000/openai/models ``` ## Related Guides - [How to deploy with OpenAI-compatible API](../deployment/openai-api/) - Complete deployment guide with authentication examples - [How to deploy as an MCP server](../deployment/mcp-server/) - MCP deployment with authentication setup - [API Reference - Server Adapters](../../reference/api/adapters/) - Technical details about authentication implementation # How to install Xaibo with different dependency groups This guide shows you how to install Xaibo with only the dependencies you need for your specific use case. ## Prerequisites Before installing Xaibo, ensure you have: - Python 3.10 or higher installed - pip or uv package manager ## Install core Xaibo package Install the minimal Xaibo package without optional dependencies: ```bash pip install xaibo ``` This gives you the core framework with basic functionality. ## Install with specific LLM providers ### OpenAI models (GPT-3.5, GPT-4) ```bash pip install xaibo[openai] ``` ### Anthropic Claude models ```bash pip install xaibo[anthropic] ``` ### Google Gemini models ```bash pip install xaibo[google] ``` ### AWS Bedrock models ```bash pip install xaibo[bedrock] ``` ### Multiple LLM providers ```bash pip install xaibo[openai,anthropic,google] ``` ## Install with web server capabilities For running the debug UI and API endpoints: ```bash pip install xaibo[webserver] ``` This includes FastAPI, Strawberry GraphQL, and other web dependencies. ## Install with local AI capabilities For local embeddings, tokenization, and transformers: ```bash pip install xaibo[local] ``` This includes sentence-transformers, tiktoken, and Hugging Face transformers. ## Install everything For full functionality with all optional dependencies: ```bash pip install xaibo[webserver,openai,anthropic,google,bedrock,local] ``` ## Install for development If you're contributing to Xaibo or need development tools: ```bash pip install xaibo[dev] ``` This includes coverage tools and development utilities. ## Using uv (recommended) For faster installation and better dependency resolution, use [`uv`](https://docs.astral.sh/uv/): ```bash # Install uv first pip install uv # Install Xaibo with uv uv add xaibo[openai,webserver] ``` ## Verify installation Test your installation by checking the version: ```bash python -c "import xaibo; print(xaibo.__version__)" ``` Or use the CLI: ```bash xaibo --version ``` ## Quick start after installation Initialize a new project to test your installation: ```bash # Create a new project uvx xaibo init my_test_project # Navigate to the project cd my_test_project # Start the development server uv run xaibo dev ``` Visit `http://localhost:9001` to see the debug UI and confirm everything is working. ## Troubleshooting ### Import errors for optional dependencies If you get import errors like `ModuleNotFoundError: No module named 'openai'`, install the missing dependency group: ```bash pip install xaibo[openai] ``` ### Version conflicts If you encounter dependency conflicts, try using uv which has better dependency resolution: ```bash uv add xaibo[your,dependencies,here] ``` ### Virtual environment issues Always install Xaibo in a virtual environment to avoid conflicts: ```bash python -m venv xaibo-env source xaibo-env/bin/activate # On Windows: xaibo-env\Scripts\activate pip install xaibo[your,dependencies] ``` # How to Start the OpenAI Responses API Get the OpenAI Responses API running in 2 minutes with the simplest possible setup. ## Prerequisites You need: - Xaibo installed with webserver support: `pip install xaibo[webserver]` - At least one agent configuration in an [`./agents/`](../../../tutorial/getting-started/) directory ## Start the Server Run this single command: ```bash python -m xaibo.server.web --adapter xaibo.server.adapters.openai_responses.OpenAiResponsesApiAdapter ``` The server starts on [`http://localhost:8000`](http://localhost:8000) and automatically: - Loads all agents from [`./agents/`](../../../tutorial/getting-started/) - Creates a `./responses/` directory for response storage - Initializes a SQLite database for persistence ## Test the API Make your first request: ```bash curl -X POST http://localhost:8000/openai/responses \ -H "Content-Type: application/json" \ -d '{ "input": "Hello, how are you today?", "model": "example" }' ``` Replace `"example"` with your actual agent ID from your [`agents/`](../../../tutorial/getting-started/) directory. ### Expected Response You'll get a JSON response like this: ```json { "id": "resp_abc123", "object": "response", "status": "completed", "model": "example", "output": [ { "type": "message", "role": "assistant", "content": [ { "type": "output_text", "text": "Hello! I'm doing well, thank you for asking...", "annotations": [] } ] } ], "usage": { "input_tokens": 12, "output_tokens": 18, "total_tokens": 30 } } ``` ## What's Next Your API is now running! Here are your next steps: - **API Reference**: See the complete [OpenAI Responses Adapter documentation](../../../reference/api/openai-responses-adapter/) for all endpoints, parameters, and response formats - **Advanced Deployment**: Learn about [OpenAI-compatible API deployment](../../deployment/openai-api/) for production setups - **Agent Configuration**: Review [agent configuration](../../../reference/agent-config/) to customize your agents - **Troubleshooting**: Check the [troubleshooting guide](../../../reference/troubleshooting/) if you encounter issues # How to deploy as an MCP server This guide shows you how to deploy your Xaibo agents as an MCP (Model Context Protocol) server, making them available as tools for other MCP-compatible applications and development environments. ## Install web server dependencies Install the required web server dependencies: ```bash pip install xaibo[webserver] ``` This includes the MCP adapter and JSON-RPC 2.0 support. ## Deploy using the CLI Use the built-in CLI for quick MCP deployment: ```bash # Deploy as MCP server (no authentication) python -m xaibo.server.web \ --agent-dir ./agents \ --adapter xaibo.server.adapters.McpApiAdapter \ --host 127.0.0.1 \ --port 8000 # Deploy with API key authentication python -m xaibo.server.web \ --agent-dir ./agents \ --adapter xaibo.server.adapters.McpApiAdapter \ --host 127.0.0.1 \ --port 8000 \ --mcp-api-key your-secret-key-here ``` ## Create an MCP server deployment Create a deployment script for your MCP server: ```python # mcp_deploy.py from xaibo import Xaibo from xaibo.server.web import XaiboWebServer def main(): # Initialize Xaibo xaibo = Xaibo() # Create web server with MCP adapter and API key server = XaiboWebServer( xaibo=xaibo, adapters=["xaibo.server.adapters.McpApiAdapter"], agent_dir="./agents", host="127.0.0.1", port=8000, mcp_api_key="your-secret-key-here" # Optional authentication ) # Start the server server.start() if __name__ == "__main__": main() ``` ## Test your MCP server Test the MCP server with curl: ### Without Authentication ```bash # Initialize MCP connection curl -X POST http://127.0.0.1:8000/mcp \ -H "Content-Type: application/json" \ -d '{ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "clientInfo": { "name": "test-client", "version": "1.0.0" } } }' # List available tools (your agents) curl -X POST http://127.0.0.1:8000/mcp \ -H "Content-Type: application/json" \ -d '{ "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} }' # Call an agent curl -X POST http://127.0.0.1:8000/mcp \ -H "Content-Type: application/json" \ -d '{ "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "your-agent-id", "arguments": { "message": "Hello from MCP client!" } } }' ``` ### With Authentication ```bash # Initialize MCP connection with API key curl -X POST http://127.0.0.1:8000/mcp \ -H "Content-Type: application/json" \ -H "Authorization: Bearer your-secret-key-here" \ -d '{ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "clientInfo": { "name": "test-client", "version": "1.0.0" } } }' # List available tools with API key curl -X POST http://127.0.0.1:8000/mcp \ -H "Content-Type: application/json" \ -H "Authorization: Bearer your-secret-key-here" \ -d '{ "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} }' # Call an agent with API key curl -X POST http://127.0.0.1:8000/mcp \ -H "Content-Type: application/json" \ -H "Authorization: Bearer your-secret-key-here" \ -d '{ "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "your-agent-id", "arguments": { "message": "Hello from MCP client!" } } }' ``` ## Use with MCP clients Connect your MCP server to various MCP clients: ### Claude Desktop #### Without Authentication Add to your Claude Desktop MCP configuration: ```json { "mcpServers": { "xaibo-agents": { "command": "python", "args": ["-m", "xaibo.server.web", "--agent-dir", "./agents", "--adapter", "xaibo.server.adapters.McpApiAdapter"], "env": { "OPENAI_API_KEY": "your-key-here" } } } } ``` #### With Authentication ```json { "mcpServers": { "xaibo-agents": { "command": "python", "args": [ "-m", "xaibo.server.web", "--agent-dir", "./agents", "--adapter", "xaibo.server.adapters.McpApiAdapter", "--mcp-api-key", "your-secret-key-here" ], "headers": { "Authorization": "Bearer your-secret-key-here" } } } } ``` ### Cline (VS Code Extension) #### Without Authentication Configure Cline to use your Xaibo MCP server: ```json { "mcp": { "servers": [ { "name": "xaibo-agents", "transport": { "type": "stdio", "command": "python", "args": ["-m", "xaibo.server.web", "--agent-dir", "./agents", "--adapter", "xaibo.server.adapters.McpApiAdapter"] } } ] } } ``` ## Deploy as stdio MCP server Create a stdio-based MCP server for direct process communication: ```python # stdio_mcp_server.py import sys import json import asyncio from xaibo import Xaibo from xaibo.server.adapters.mcp import McpApiAdapter class StdioMCPServer: def __init__(self): self.xaibo = Xaibo() self.xaibo.register_agents_from_directory("./agents") self.adapter = McpApiAdapter(self.xaibo) async def handle_request(self, request_data): """Handle a single MCP request""" try: request = json.loads(request_data) response = await self.adapter.handle_mcp_request(request) return json.dumps(response) except Exception as e: error_response = { "jsonrpc": "2.0", "id": request.get("id") if "request" in locals() else None, "error": { "code": -32603, "message": f"Internal error: {str(e)}" } } return json.dumps(error_response) async def run(self): """Run the stdio MCP server""" while True: try: # Read from stdin line = await asyncio.get_event_loop().run_in_executor( None, sys.stdin.readline ) if not line: break # Process request response = await self.handle_request(line.strip()) # Write to stdout print(response, flush=True) except KeyboardInterrupt: break except Exception as e: error_response = { "jsonrpc": "2.0", "id": None, "error": { "code": -32603, "message": f"Server error: {str(e)}" } } print(json.dumps(error_response), flush=True) async def main(): server = StdioMCPServer() await server.run() if __name__ == "__main__": asyncio.run(main()) ``` Use the stdio server: ```bash # Run as stdio server python stdio_mcp_server.py # Or use with MCP client that expects stdio echo '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}' | python stdio_mcp_server.py ``` ## API Key Configuration Detailed Authentication Setup For comprehensive authentication setup including troubleshooting and security best practices, see the [authentication guide](../../authentication/). ### Environment Variables The recommended approach is to use environment variables for API key configuration: ```bash # Set environment variable export MCP_API_KEY="your-secret-key-here" # Start server (API key automatically detected) python -m xaibo.server.web \ --agent-dir ./agents \ --adapter xaibo.server.adapters.McpApiAdapter \ --host 127.0.0.1 \ --port 8000 ``` ### Command Line Arguments Alternatively, provide API keys via command line arguments: ```bash # API key via command line (overrides environment variable) python -m xaibo.server.web \ --agent-dir ./agents \ --adapter xaibo.server.adapters.McpApiAdapter \ --host 127.0.0.1 \ --port 8000 \ --mcp-api-key your-secret-key-here ``` ### Docker Deployment ```dockerfile # Dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . # Use environment variable for API key ENV MCP_API_KEY="" CMD ["python", "-m", "xaibo.server.web", \ "--agent-dir", "./agents", \ "--adapter", "xaibo.server.adapters.McpApiAdapter", \ "--host", "0.0.0.0", \ "--port", "8000"] ``` ```bash # Run with Docker docker run -e MCP_API_KEY=your-secret-key-here -p 8000:8000 your-xaibo-image ``` ## Best practices ### Agent design for MCP - Keep agent responses concise and focused - Design tools that work well as discrete operations - Use clear, descriptive agent and tool names - Implement proper error handling ### Security - Validate all input parameters - Implement rate limiting for production - Always use authentication for production environments - Use environment variables for secrets - Rotate API keys regularly - Monitor for unusual usage patterns ### Performance - Optimize agent response times - Cache frequently used data - Use connection pooling for external services - Monitor memory usage ### Reliability - Implement proper error handling - Add health checks and monitoring - Use graceful shutdown procedures - Log all operations for debugging ## Troubleshooting ### MCP protocol errors - Verify JSON-RPC 2.0 format compliance - Check method names and parameter structures - Review MCP specification for requirements - Test with simple MCP clients first ### Agent execution errors - Check agent configurations and dependencies - Verify tool implementations work correctly - Review logs for detailed error messages - Test agents independently before MCP integration ### Connection issues - Verify server is listening on correct port - Check firewall and network settings - Test with curl before using MCP clients - Review client configuration and compatibility ### Performance problems - Monitor response times and resource usage - Optimize agent configurations - Check for memory leaks in long-running servers - Use profiling tools to identify bottlenecks # How to deploy with OpenAI-compatible API This guide shows you how to deploy your Xaibo agents as an OpenAI-compatible REST API, allowing any OpenAI client to interact with your agents. Available OpenAI Adapters Xaibo provides two OpenAI-compatible adapters: - **[OpenAI API Adapter](../../../reference/api/adapters/#openaiapiadapter)** - Basic OpenAI Chat Completions API compatibility (covered in this guide) - **[OpenAI Responses Adapter](../../../reference/api/openai-responses-adapter/)** - Advanced response management with conversation history and persistence ## Install web server dependencies Install the required web server dependencies: ```bash pip install xaibo[webserver] ``` This includes FastAPI, Strawberry GraphQL, and other web server components. ## Deploy using the CLI Use the built-in CLI for quick deployment: ```bash # Deploy all agents in the agents directory (no authentication) python -m xaibo.server.web \ --agent-dir ./agents \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --host 0.0.0.0 \ --port 8000 # Deploy with API key authentication python -m xaibo.server.web \ --agent-dir ./agents \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --host 0.0.0.0 \ --port 8000 \ --openai-api-key sk-your-secret-key-here ``` ## Test your deployment Test the deployed API with curl: ### Without Authentication ```bash # List available models (agents) curl -X GET http://localhost:8000/openai/models # Send a chat completion request curl -X POST http://localhost:8000/openai/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "your-agent-id", "messages": [ {"role": "user", "content": "Hello, how are you?"} ], "temperature": 0.7, "max_tokens": 1000 }' # Test streaming responses curl -X POST http://localhost:8000/openai/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "your-agent-id", "messages": [ {"role": "user", "content": "Tell me a story"} ], "stream": true }' ``` ### With Authentication ```bash # List available models (agents) with API key curl -X GET http://localhost:8000/openai/models \ -H "Authorization: Bearer sk-your-secret-key-here" # Send a chat completion request with API key curl -X POST http://localhost:8000/openai/chat/completions \ -H "Content-Type: application/json" \ -H "Authorization: Bearer sk-your-secret-key-here" \ -d '{ "model": "your-agent-id", "messages": [ {"role": "user", "content": "Hello, how are you?"} ], "temperature": 0.7, "max_tokens": 1000 }' # Test streaming responses with API key curl -X POST http://localhost:8000/openai/chat/completions \ -H "Content-Type: application/json" \ -H "Authorization: Bearer sk-your-secret-key-here" \ -d '{ "model": "your-agent-id", "messages": [ {"role": "user", "content": "Tell me a story"} ], "stream": true }' ``` ## Use with OpenAI client libraries Connect using official OpenAI client libraries: ### Python client #### Without Authentication ```python # client_test.py from openai import OpenAI # Point to your Xaibo deployment (no authentication) client = OpenAI( base_url="http://localhost:8000/openai", api_key="not-needed" # Xaibo doesn't require API key by default ) # List available models (your agents) models = client.models.list() print("Available agents:") for model in models.data: print(f" - {model.id}") # Chat with an agent response = client.chat.completions.create( model="your-agent-id", messages=[ {"role": "user", "content": "What can you help me with?"} ] ) API_ print(response.choices[0].message.content) ``` #### With Authentication ```python # client_test_auth.py from openai import OpenAI # Point to your Xaibo deployment (with authentication) client = OpenAI( base_url="http://localhost:8000/openai", api_key="sk-your-secret-key-here" # Use your configured API key ) # List available models (your agents) models = client.models.list() print("Available agents:") for model in models.data: print(f" - {model.id}") # Chat with an agent response = client.chat.completions.create( model="your-agent-id", messages=[ {"role": "user", "content": "What can you help me with?"} ] ) print(response.choices[0].message.content) ``` ### Node.js client #### Without Authentication ```javascript // client_test.js import OpenAI from 'openai'; const openai = new OpenAI({ baseURL: 'http://localhost:8000/openai', apiKey: 'not-needed' }); async function testAgent() { // List available models const models = await openai.models.list(); console.log('Available agents:', models.data.map(m => m.id)); // Chat with an agent const completion = await openai.chat.completions.create({ model: 'your-agent-id', messages: [ { role: 'user', content: 'Hello from Node.js!' } ] }); console.log(completion.choices[0].message.content); } testAgent().catch(console.error); ``` #### With Authentication ```javascript // client_test_auth.js import OpenAI from 'openai'; const openai = new OpenAI({ baseURL: 'http://localhost:8000/openai', apiKey: 'sk-your-secret-key-here' // Use your configured API key }); async function testAgent() { // List available models const models = await openai.models.list(); console.log('Available agents:', models.data.map(m => m.id)); // Chat with an agent const completion = await openai.chat.completions.create({ model: 'your-agent-id', messages: [ { role: 'user', content: 'Hello from Node.js!' } ] }); console.log(completion.choices[0].message.content); } testAgent().catch(console.error); ``` ## API Key Configuration Detailed Authentication Setup For comprehensive authentication setup including troubleshooting and security best practices, see the [authentication guide](../../authentication/). ### Environment Variables The recommended approach is to use environment variables for API key configuration: ```bash # Set environment variable export CUSTOM_OPENAI_API_KEY="sk-your-secret-key-here" # Start server (API key automatically detected) python -m xaibo.server.web \ --agent-dir ./agents \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --host 0.0.0.0 \ --port 8000 ``` ### Command Line Arguments Alternatively, provide API keys via command line arguments: ```bash # API key via command line (overrides environment variable) python -m xaibo.server.web \ --agent-dir ./agents \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --host 0.0.0.0 \ --port 8000 \ --openai-api-key sk-your-secret-key-here ``` ### Docker Deployment ```dockerfile # Dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . # Use environment variable for API key ENV CUSTOM_OPENAI_API_KEY="" CMD ["python", "-m", "xaibo.server.web", \ "--agent-dir", "./agents", \ "--adapter", "xaibo.server.adapters.OpenAiApiAdapter", \ "--host", "0.0.0.0", \ "--port", "8000"] ``` ```bash # Run with Docker docker run -e CUSTOM_OPENAI_API_KEY=sk-your-secret-key-here -p 8000:8000 your-xaibo-image ``` ## Best practices ### Security - Use HTTPS in production - Always use API key authentication in production environments - Set up proper CORS policies - Use environment variables for secrets - Rotate API keys regularly - Monitor API key usage for unusual patterns ### Performance - Use multiple workers for high load - Implement connection pooling - Cache agent configurations - Monitor resource usage ### Reliability - Add health checks and monitoring - Implement graceful shutdown - Use load balancers for redundancy - Set up automated restarts ### Scaling - Use container orchestration (Kubernetes, ECS) - Implement horizontal pod autoscaling - Monitor and optimize resource usage - Use CDN for static assets ## Troubleshooting ### Port binding issues - Check if port is already in use - Verify firewall settings - Use different port numbers for testing ### Agent loading errors - Verify agent YAML syntax - Check that all dependencies are installed - Review agent configuration paths ### Performance problems - Monitor CPU and memory usage - Optimize agent configurations - Use profiling tools to identify bottlenecks - Consider scaling horizontally ### Connection issues - Verify network connectivity - Check DNS resolution - Review proxy and load balancer settings - Test with simple HTTP clients first # How to Use Xaibo Agents in LiveKit Voice Assistants This guide shows you how to integrate your Xaibo agents with LiveKit's voice assistant framework to create real-time conversational AI applications. *See also: [LiveKit Integration Reference](../../../reference/integrations/livekit/)* Complete Working Example For a ready-to-run implementation, see the [LiveKit Example](https://github.com/xpressai/xaibo/tree/main/examples/livekit_example) which includes: - Complete [`agent.py`](https://github.com/xpressai/xaibo/blob/main/examples/livekit_example/agent.py) implementation - Environment configuration template ([`.env`](https://github.com/xpressai/xaibo/blob/main/examples/livekit_example/.env)) - Project dependencies ([`pyproject.toml`](https://github.com/xpressai/xaibo/blob/main/examples/livekit_example/pyproject.toml)) - Detailed setup and usage instructions ([`README.md`](https://github.com/xpressai/xaibo/blob/main/examples/livekit_example/README.md)) ## Prerequisites - Python 3.10 or higher - uv package manager installed - Agent configurations in YAML format - Environment variables configured (`.env` file) ## Install Dependencies Install Xaibo with LiveKit integration support: ```bash uv add xaibo[livekit] ``` Install LiveKit agents framework and speech plugins: ```bash uv add livekit-agents livekit-plugins-openai livekit-plugins-silero ``` ## Set Up Your LiveKit Worker Create a new Python file for your LiveKit worker (e.g., `voice_assistant.py`): ```python import logging from dotenv import load_dotenv from livekit import rtc from livekit.agents import ( AutoSubscribe, JobContext, WorkerOptions, cli, Agent, AgentSession ) from livekit.plugins import openai, silero from xaibo.integrations.livekit import XaiboAgentLoader # Load environment variables load_dotenv(dotenv_path=".env") # Configure logging logger = logging.getLogger("xaibo-voice-assistant") logger.setLevel(logging.INFO) ``` ## Load Your Xaibo Agents Use the [`XaiboAgentLoader`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py) to load your agent configurations: ```python # Initialize the agent loader loader = XaiboAgentLoader() # Load all agents from your configuration directory loader.load_agents_from_directory("./agents") # Enable debug logging (optional) loader.enable_debug_logging("./debug") ``` ## Create Your Voice Assistant Get an LLM instance from your loaded agent and configure the LiveKit assistant: ```python # Get the LLM for your specific agent llm = loader.get_llm("your-agent-id") # Create the voice assistant with speech components assistant = Agent( instructions="", # Instructions handled by Xaibo agent vad=silero.VAD.load(), # Voice Activity Detection stt=openai.STT(), # Speech-to-Text llm=llm, # Your Xaibo agent as LLM tts=openai.TTS(), # Text-to-Speech ) ``` ## Implement the LiveKit Entrypoint Create the main entrypoint function that LiveKit will call: ```python async def entrypoint(ctx: JobContext): logger.info(f"Connecting to room {ctx.room.name}") # Connect to the LiveKit room await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY) # Start the agent session session = AgentSession() await session.start( agent=assistant, room=ctx.room, ) logger.info("Voice assistant started") ``` ## Run Your Voice Assistant Add the main block to run your LiveKit worker: ```python if __name__ == "__main__": cli.run_app( WorkerOptions( entrypoint_fnc=entrypoint, ) ) ``` ## Complete Example Here's the complete working example: ```python import logging from dotenv import load_dotenv from livekit import rtc from livekit.agents import ( AutoSubscribe, JobContext, WorkerOptions, cli, Agent, AgentSession ) from livekit.plugins import openai, silero from xaibo.integrations.livekit import XaiboAgentLoader # Load environment variables load_dotenv(dotenv_path=".env") # Configure logging logger = logging.getLogger("xaibo-voice-assistant") logger.setLevel(logging.INFO) # Load Xaibo agents loader = XaiboAgentLoader() loader.load_agents_from_directory("./agents") loader.enable_debug_logging("./debug") # Get LLM from your agent llm = loader.get_llm("example") # Create voice assistant assistant = Agent( instructions="", vad=silero.VAD.load(), stt=openai.STT(), llm=llm, tts=openai.TTS(), ) async def entrypoint(ctx: JobContext): logger.info(f"Connecting to room {ctx.room.name}") await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY) session = AgentSession() await session.start( agent=assistant, room=ctx.room, ) logger.info("Voice assistant started") if __name__ == "__main__": cli.run_app( WorkerOptions( entrypoint_fnc=entrypoint, ) ) ``` ## Run Your Voice Assistant Execute your voice assistant worker: ```bash python voice_assistant.py dev ``` ## Advanced Configuration ### Enable File Watching Automatically reload agent configurations when files change: ```python loader.enable_file_watching("./agents") ``` ### List Available Agents Check which agents are loaded: ```python available_agents = loader.list_agents() print(f"Available agents: {available_agents}") ``` ### Get Agent Information Retrieve detailed information about a specific agent: ```python agent_info = loader.get_agent_info("your-agent-id") print(f"Agent modules: {agent_info['modules']}") ``` ### Custom Debug Directory Specify a custom directory for debug traces: ```python loader.enable_debug_logging("./custom-debug-path") ``` ## Environment Variables Ensure your `.env` file contains the necessary API keys: ```text OPENAI_API_KEY=your_openai_api_key LIVEKIT_URL=your_livekit_server_url LIVEKIT_API_KEY=your_livekit_api_key LIVEKIT_API_SECRET=your_livekit_api_secret ``` ## Troubleshooting ### Agent Not Found Error If you get an "Agent not found" error: 1. Verify your agent YAML files are in the correct directory 1. Check that the agent ID matches what's in your YAML configuration 1. Use [`loader.list_agents()`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py:97) to see available agents ### Debug Logging Issues If debug logging fails to enable: 1. Ensure you have the UI dependencies installed 1. Check that the debug directory is writable 1. Verify the debug directory path exists or can be created ### Connection Issues For LiveKit connection problems: 1. Verify your LiveKit server URL and credentials 1. Check network connectivity to your LiveKit server 1. Ensure your room name is valid and accessible # How to switch between different LLM providers This guide shows you how to configure your Xaibo agents to use different LLM providers like OpenAI, Anthropic, Google, or AWS Bedrock. ## Switch to OpenAI models Install the OpenAI dependency and configure your agent: ```bash pip install xaibo[openai] ``` ```yaml # agents/openai_agent.yml id: openai-agent modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: llm config: model: gpt-4 api_key: ${OPENAI_API_KEY} # Optional, uses env var by default temperature: 0.7 max_tokens: 2000 - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator ``` Set your API key: ```bash export OPENAI_API_KEY=your_openai_api_key_here ``` ## Switch to Anthropic Claude Install Anthropic dependencies and configure Claude: ```bash pip install xaibo[anthropic] ``` ```yaml # agents/claude_agent.yml id: claude-agent modules: - module: xaibo.primitives.modules.llm.AnthropicLLM id: llm config: model: claude-3-5-sonnet-20241022 api_key: ${ANTHROPIC_API_KEY} temperature: 0.7 max_tokens: 4000 - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator ``` Set your API key: ```bash export ANTHROPIC_API_KEY=your_anthropic_api_key_here ``` ## Switch to Google Gemini Install Google dependencies and configure Gemini: ```bash pip install xaibo[google] ``` ```yaml # agents/gemini_agent.yml id: gemini-agent modules: - module: xaibo.primitives.modules.llm.GoogleLLM id: llm config: model: gemini-2.0-flash-001 api_key: ${GOOGLE_API_KEY} temperature: 0.7 max_tokens: 2000 - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator ``` Set your API key: ```bash export GOOGLE_API_KEY=your_google_api_key_here ``` ## Switch to AWS Bedrock Install Bedrock dependencies and configure AWS models: ```bash pip install xaibo[bedrock] ``` ```yaml # agents/bedrock_agent.yml id: bedrock-agent modules: - module: xaibo.primitives.modules.llm.BedrockLLM id: llm config: model: anthropic.claude-3-5-sonnet-20241022-v2:0 region_name: us-east-1 aws_access_key_id: ${AWS_ACCESS_KEY_ID} aws_secret_access_key: ${AWS_SECRET_ACCESS_KEY} temperature: 0.7 max_tokens: 4000 - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator ``` Set your AWS credentials: ```bash export AWS_ACCESS_KEY_ID=your_aws_access_key export AWS_SECRET_ACCESS_KEY=your_aws_secret_key export AWS_DEFAULT_REGION=us-east-1 ``` ## Use custom API endpoints Configure custom endpoints for OpenAI-compatible APIs: ```yaml # agents/custom_openai_agent.yml id: custom-openai-agent modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: llm config: model: your-custom-model base_url: https://your-custom-endpoint.com/v1 api_key: ${CUSTOM_API_KEY} timeout: 120.0 - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator ``` This works with: - Local LLM servers (Ollama, LM Studio) - Azure OpenAI Service - Custom OpenAI-compatible APIs ## Best practices ### Provider selection - Choose based on your specific use case requirements - Consider cost, performance, and feature needs - Test multiple providers for critical applications ### Configuration management - Use environment variables for API keys - Store provider configs in separate files - Version control your configurations ### Fallback strategies - Configure backup providers for reliability - Implement retry logic with exponential backoff - Monitor provider availability and performance ### Security - Rotate API keys regularly - Use least-privilege access policies - Monitor usage for anomalies ## Troubleshooting ### Authentication errors - Verify API keys are correct and active - Check account billing status and limits - Ensure proper environment variable setup ### Model availability - Verify model names match provider specifications - Check regional availability for specific models - Update to latest model versions when needed ### Rate limiting - Implement proper retry logic with backoff - Monitor usage against provider limits - Consider upgrading to higher-tier plans ### Performance issues - Adjust timeout values for slow responses - Optimize prompt length and complexity - Monitor token usage and costs # How to set up vector memory for agents This guide shows you how to enable your Xaibo agents to store and retrieve information using vector embeddings, allowing them to remember and search through previous conversations and documents. ## Install memory dependencies Install the required dependencies for local embeddings: ```bash pip install xaibo[local] ``` This includes sentence-transformers, tiktoken, and other memory-related packages. ## Configure basic vector memory Add vector memory to your agent configuration: ```yaml # agents/memory_agent.yml id: memory-agent description: An agent with vector memory capabilities modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: llm config: model: gpt-4.1-nano # Text chunker for splitting documents - module: xaibo.primitives.modules.memory.TokenChunker id: chunker config: window_size: 512 window_overlap: 50 encoding_name: "cl100k_base" # Embedder for converting text to vectors - module: xaibo.primitives.modules.memory.SentenceTransformerEmbedder id: embedder config: model_name: "all-MiniLM-L6-v2" # Vector index for storage and retrieval - module: xaibo.primitives.modules.memory.NumpyVectorIndex id: vector_index config: storage_dir: "./memory_storage" # Main vector memory module - module: xaibo.primitives.modules.memory.VectorMemory id: memory config: memory_file_path: "./agent_memory.pkl" - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator config: system_prompt: | You are a helpful assistant with memory capabilities. You can remember information from previous conversations. exchange: # Connect memory components - module: memory protocol: ChunkerProtocol provider: chunker - module: memory protocol: EmbedderProtocol provider: embedder - module: memory protocol: VectorIndexProtocol provider: vector_index - module: orchestrator protocol: MemoryProtocol provider: memory ``` ## Use OpenAI embeddings Configure OpenAI embeddings for higher quality vectors: ```bash pip install xaibo[openai] ``` ```yaml # Replace the embedder module with OpenAI modules: - module: xaibo.primitives.modules.memory.OpenAIEmbedder id: embedder config: model: "text-embedding-3-small" api_key: ${OPENAI_API_KEY} dimensions: 1536 ``` Set your API key: ```bash export OPENAI_API_KEY=your_openai_api_key_here ``` ## Configure Hugging Face embeddings Use Hugging Face models for embeddings: ```yaml modules: - module: xaibo.primitives.modules.memory.HuggingFaceEmbedder id: embedder config: model_name: "sentence-transformers/all-mpnet-base-v2" device: "cuda" # Use "cpu" if no GPU available max_length: 512 pooling_strategy: "mean" ``` Popular Hugging Face embedding models: - `sentence-transformers/all-mpnet-base-v2` - High quality, balanced - `sentence-transformers/all-MiniLM-L6-v2` - Fast and lightweight - `sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2` - Multilingual support ## Customize text chunking Configure chunking strategy for your content: ```yaml modules: - module: xaibo.primitives.modules.memory.TokenChunker id: chunker config: window_size: 1024 # Larger chunks for documents window_overlap: 100 # More overlap for context encoding_name: "cl100k_base" # GPT-4 tokenizer ``` Chunking strategies: - **Small chunks (256-512 tokens)**: Better for precise retrieval - **Medium chunks (512-1024 tokens)**: Balanced approach - **Large chunks (1024-2048 tokens)**: Better context preservation ## Store memory persistently Configure persistent storage for your vector memory: ```yaml modules: - module: xaibo.primitives.modules.memory.NumpyVectorIndex id: vector_index config: storage_dir: "/path/to/persistent/storage" - module: xaibo.primitives.modules.memory.VectorMemory id: memory config: memory_file_path: "/path/to/persistent/memory.pkl" ``` Create the storage directory: ```bash mkdir -p /path/to/persistent/storage ``` ## Best practices ### Embedding model selection - Use OpenAI embeddings for highest quality - Use local models for privacy and cost control - Choose model size based on performance requirements ### Chunking strategy - Smaller chunks for precise retrieval - Larger chunks for better context - Adjust overlap based on content type ### Storage management - Use persistent storage for production - Monitor storage size and performance - Implement cleanup strategies for old data ### Performance optimization - Use GPU acceleration when available - Cache frequently accessed vectors - Batch process large document collections ## Troubleshooting ### Memory not persisting - Check file permissions for storage directories - Verify storage paths are absolute and accessible - Ensure sufficient disk space ### Poor retrieval quality - Experiment with different embedding models - Adjust similarity thresholds - Review chunking strategy for your content ### Performance issues - Monitor memory usage and optimize chunk sizes - Use faster embedding models for real-time applications - Consider GPU acceleration for large collections ### Import errors - Verify all memory dependencies are installed - Check that storage directories exist - Ensure proper module configuration in agent YAML # How to integrate MCP (Model Context Protocol) tools This guide shows you how to connect external MCP servers to your Xaibo agents, giving them access to tools from other applications and services. ## Configure MCP tool provider Add the MCP tool provider to your agent configuration: ```yaml # agents/mcp_agent.yml id: mcp-agent description: An agent with MCP tools modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: llm config: model: gpt-4.1-nano - id: mcp-tools module: xaibo.primitives.modules.tools.MCPToolProvider config: timeout: 60.0 servers: - name: filesystem transport: stdio command: ["python", "-m", "mcp_server_filesystem"] args: ["--root", "/workspace"] - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator config: max_thoughts: 10 system_prompt: | You are a helpful assistant with access to filesystem tools. You can read, write, and manage files through MCP tools. ``` ## Connect to stdio MCP servers Configure local MCP servers that run as separate processes: ```yaml servers: # Filesystem server - name: filesystem transport: stdio command: ["python", "-m", "mcp_server_filesystem"] args: ["--root", "/workspace"] env: LOG_LEVEL: "INFO" # Git server - name: git transport: stdio command: ["npx", "@modelcontextprotocol/server-git"] args: ["--repository", "/path/to/repo"] # SQLite server - name: database transport: stdio command: ["python", "-m", "mcp_server_sqlite"] args: ["--db-path", "/path/to/database.db"] ``` ## Connect to HTTP-based MCP servers Configure MCP servers accessible over HTTP using Server-Sent Events: ```yaml servers: # Web search server - name: web_search transport: sse url: "https://api.example.com/mcp" headers: Authorization: "Bearer your-api-key" Content-Type: "application/json" # Custom API server - name: custom_api transport: sse url: "https://your-mcp-server.com/mcp" headers: X-API-Key: "your-api-key" User-Agent: "Xaibo-Agent/1.0" ``` ## Connect to WebSocket MCP servers Configure MCP servers that use WebSocket connections: ```yaml servers: # Real-time data server - name: realtime_data transport: websocket url: "ws://localhost:8080/mcp" headers: X-API-Key: "your-websocket-key" # Chat server - name: chat_server transport: websocket url: "wss://chat.example.com/mcp" headers: Authorization: "Bearer your-token" ``` ## Use multiple MCP servers Configure multiple MCP servers in a single agent: ```yaml # agents/multi_mcp_agent.yml id: multi-mcp-agent modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: llm config: model: gpt-4 - id: mcp-tools module: xaibo.primitives.modules.tools.MCPToolProvider config: timeout: 30.0 servers: # Local filesystem access - name: fs transport: stdio command: ["python", "-m", "mcp_server_filesystem"] args: ["--root", "."] # Git repository management - name: git transport: stdio command: ["npx", "@modelcontextprotocol/server-git"] args: ["--repository", "."] # Web search capabilities - name: search transport: sse url: "https://search-api.example.com/mcp" headers: Authorization: "Bearer your-search-api-key" # Database operations - name: db transport: websocket url: "ws://localhost:8080/mcp" headers: X-Database-Key: "your-db-key" - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator config: system_prompt: | You are a development assistant with access to: - Filesystem operations (fs.* tools) - Git repository management (git.* tools) - Web search capabilities (search.* tools) - Database operations (db.* tools) Use these tools to help with development tasks. ``` ## Test MCP tool integration Start your agent and verify MCP tools are available: ```bash # Start the development server uv run xaibo dev # Test with a filesystem operation curl -X POST http://127.0.0.1:9001/openai/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "mcp-agent", "messages": [ {"role": "user", "content": "List the files in the current directory"} ] }' ``` ## Install common MCP servers Install popular MCP servers for common use cases: ```bash # Filesystem server pip install mcp-server-filesystem # Git server (Node.js) npm install -g @modelcontextprotocol/server-git # SQLite server pip install mcp-server-sqlite # GitHub server npm install -g @modelcontextprotocol/server-github # Brave search server npm install -g @modelcontextprotocol/server-brave-search ``` ## Create a custom MCP server Build your own MCP server for custom functionality: ```python # custom_mcp_server.py import asyncio import json from mcp.server import Server from mcp.types import Tool, TextContent app = Server("custom-tools") @app.list_tools() async def list_tools(): return [ Tool( name="calculate_fibonacci", description="Calculate the nth Fibonacci number", inputSchema={ "type": "object", "properties": { "n": { "type": "integer", "description": "The position in the Fibonacci sequence" } }, "required": ["n"] } ), Tool( name="reverse_string", description="Reverse a string", inputSchema={ "type": "object", "properties": { "text": { "type": "string", "description": "The string to reverse" } }, "required": ["text"] } ) ] @app.call_tool() async def call_tool(name: str, arguments: dict): if name == "calculate_fibonacci": n = arguments["n"] if n <= 0: return [TextContent(type="text", text="Error: n must be positive")] a, b = 0, 1 for _ in range(n): a, b = b, a + b return [TextContent(type="text", text=f"The {n}th Fibonacci number is {a}")] elif name == "reverse_string": text = arguments["text"] reversed_text = text[::-1] return [TextContent(type="text", text=f"Reversed: {reversed_text}")] else: return [TextContent(type="text", text=f"Unknown tool: {name}")] if __name__ == "__main__": asyncio.run(app.run()) ``` Use your custom MCP server: ```yaml servers: - name: custom transport: stdio command: ["python", "custom_mcp_server.py"] ``` ## Handle MCP server authentication Configure authentication for secure MCP servers: ```yaml servers: # API key authentication - name: secure_api transport: sse url: "https://secure-api.example.com/mcp" headers: Authorization: "Bearer ${API_TOKEN}" X-Client-ID: "${CLIENT_ID}" # OAuth token authentication - name: oauth_server transport: websocket url: "wss://oauth-server.com/mcp" headers: Authorization: "Bearer ${OAUTH_TOKEN}" # Custom authentication - name: custom_auth transport: sse url: "https://custom.example.com/mcp" headers: X-API-Key: "${CUSTOM_API_KEY}" X-Signature: "${REQUEST_SIGNATURE}" ``` Set environment variables for authentication: ```bash # .env file API_TOKEN=your_api_token_here CLIENT_ID=your_client_id OAUTH_TOKEN=your_oauth_token CUSTOM_API_KEY=your_custom_key REQUEST_SIGNATURE=your_signature ``` ## Monitor MCP connections Check MCP server status and debug connection issues: ```python # debug_mcp.py import asyncio from xaibo.primitives.modules.tools.mcp_tool_provider import MCPToolProvider async def test_mcp_connection(): config = { "servers": [ { "name": "test_server", "transport": "stdio", "command": ["python", "-m", "mcp_server_filesystem"], "args": ["--root", "."] } ] } provider = MCPToolProvider(config=config) try: # Initialize and get tools await provider.initialize() tools = await provider.get_tools() print(f"Connected to MCP server. Available tools:") for tool in tools: print(f" - {tool.name}: {tool.description}") except Exception as e: print(f"Failed to connect to MCP server: {e}") finally: await provider.cleanup() if __name__ == "__main__": asyncio.run(test_mcp_connection()) ``` ## Best practices ### Server configuration - Use descriptive server names that indicate their purpose - Set appropriate timeouts based on server response times - Group related servers logically in your configuration - Configure environment-specific settings for development, staging, and production - Document server dependencies and requirements clearly - Use consistent naming conventions across all server configurations ### Security - Store sensitive credentials in environment variables - Use HTTPS/WSS for remote connections - Validate server certificates in production - Implement proper authentication token rotation - Restrict server access to necessary network ranges - Log authentication attempts for security monitoring - Use least-privilege principles for server permissions ### Error handling - Configure fallback behavior when servers are unavailable - Monitor server health and connection status - Implement retry logic for transient failures - Design graceful degradation when tools are temporarily unavailable - Provide meaningful error messages that help users understand what went wrong - Log errors with sufficient context for debugging - Implement circuit breaker patterns for unreliable external services ### Performance - Cache tool definitions to reduce server calls - Use connection pooling for multiple requests - Monitor server response times and optimize timeouts - Implement request queuing for high-traffic scenarios - Use async operations to prevent blocking - Monitor memory usage and implement cleanup procedures - Profile tool execution times to identify bottlenecks ## Exception handling best practices ### When to use exceptions vs. error returns **Use exceptions for:** - Server connection failures and network timeouts - Protocol violations and malformed responses - Authentication and authorization failures - Critical system errors that prevent tool execution - Configuration errors that make tools unusable **Use error returns for:** - Invalid tool parameters and user input validation - Business logic errors and expected failure conditions - Resource not found scenarios - Rate limiting and quota exceeded situations - Recoverable errors that users can fix by adjusting inputs ### Structure error messages for LLM consumption Design error messages that provide clear, actionable information: - **Be specific**: Include exact parameter names, expected formats, and current values - **Provide context**: Explain what the tool was trying to accomplish when it failed - **Suggest solutions**: Offer concrete steps to resolve the issue - **Use consistent format**: Structure errors uniformly across all tools - **Include error codes**: Use standardized error types for programmatic handling - **Add examples**: Show correct parameter formats when validation fails ### Exception hierarchy and custom exception types Create a structured exception hierarchy for different error categories: - **MCPToolError**: Base exception for all tool-related errors - **ParameterValidationError**: Invalid or missing parameters - **ExternalServiceError**: Failures from external APIs or services - **ResourceNotFoundError**: Requested resources don't exist - **AuthenticationError**: Authentication or authorization failures - **TimeoutError**: Operations that exceed time limits - **ConfigurationError**: Invalid server or tool configuration ### Logging and debugging failed tool calls Implement comprehensive logging strategies: - **Structured logging**: Use consistent log formats with relevant metadata - **Call tracking**: Assign unique IDs to track tool calls across systems - **Performance metrics**: Log execution times and resource usage - **Error context**: Include full stack traces and relevant state information - **Security considerations**: Avoid logging sensitive data like credentials - **Log levels**: Use appropriate levels (DEBUG, INFO, WARN, ERROR) for different events - **Retention policies**: Configure log rotation and archival strategies ### Async tool implementations Handle asynchronous operations properly: - **Timeout management**: Set reasonable timeouts for all async operations - **Resource cleanup**: Ensure proper cleanup of connections and resources - **Cancellation handling**: Support operation cancellation when needed - **Concurrent execution**: Manage concurrent tool calls safely - **Error propagation**: Properly handle and propagate async exceptions - **Progress tracking**: Provide feedback for long-running operations ### Complex parameter validation Implement robust parameter validation: - **Type checking**: Validate parameter types before processing - **Range validation**: Check numeric values are within acceptable ranges - **Format validation**: Use regex or schemas for structured data - **Dependency validation**: Ensure related parameters are consistent - **Sanitization**: Clean and normalize input data - **Default handling**: Provide sensible defaults for optional parameters - **Custom validators**: Create reusable validation functions for complex types ### Tool testing strategies Develop comprehensive testing approaches: - **Unit tests**: Test individual tool functions in isolation - **Integration tests**: Test tool interactions with external services - **Error scenario tests**: Verify proper handling of all error conditions - **Performance tests**: Ensure tools meet response time requirements - **Security tests**: Validate input sanitization and access controls - **Mock testing**: Use mocks to test error conditions and edge cases - **End-to-end tests**: Test complete tool workflows through the MCP protocol ### Performance optimization techniques Optimize tool performance: - **Connection reuse**: Maintain persistent connections to external services - **Caching strategies**: Cache expensive computations and API responses - **Batch operations**: Group multiple operations when possible - **Lazy loading**: Load resources only when needed - **Memory management**: Monitor and optimize memory usage - **Async patterns**: Use async/await for I/O-bound operations - **Resource pooling**: Share expensive resources across tool calls ## Troubleshooting ### Server connection failures - Verify server command and arguments are correct - Check that required dependencies are installed - Ensure network connectivity for remote servers ### Authentication errors - Verify API keys and tokens are valid - Check header format matches server expectations - Ensure environment variables are properly set ### Tool execution errors - Check server logs for detailed error messages - Verify tool parameters match expected schema - Test tools directly with the MCP server before integration ### Performance issues - Increase timeout values for slow servers - Check network latency for remote connections - Monitor server resource usage and scaling ## Debug and troubleshoot tools Use Xaibo's built-in debugging capabilities to diagnose tool discovery, connection, and execution issues. ### Enable debug UI Start your development server with the debug UI enabled: ```bash # Start with debug UI (automatically enabled in dev mode) uv run xaibo dev # Access debug UI at http://localhost:9001 ``` The debug UI provides: - Real-time event tracing for all tool operations - Visual representation of tool discovery and execution - Detailed error logs and stack traces - Performance metrics for tool calls ### Use list_tools() for programmatic debugging Access tool information programmatically using the [`list_tools()`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/tools.py:9) method: ```python # debug_mcp_tools.py import asyncio from xaibo.primitives.modules.tools.mcp_tool_provider import MCPToolProvider async def debug_mcp_tools(): """Debug MCP tool discovery and availability""" config = { "servers": [ { "name": "filesystem", "transport": "stdio", "command": ["python", "-m", "mcp_server_filesystem"], "args": ["--root", "."] } ] } provider = MCPToolProvider(config=config) try: # Initialize the provider await provider.initialize() # List all available tools tools = await provider.list_tools() print(f"Found {len(tools)} tools:") for tool in tools: print(f" - {tool.name}") print(f" Description: {tool.description}") print(f" Parameters: {len(tool.parameters)} params") for param in tool.parameters: print(f" - {param.name}: {param.type} ({'required' if param.required else 'optional'})") print() except Exception as e: print(f"Error debugging tools: {e}") import traceback traceback.print_exc() finally: await provider.cleanup() if __name__ == "__main__": asyncio.run(debug_mcp_tools()) ``` ### Step-by-step debugging procedure Follow these steps to diagnose MCP tool issues: #### 1. Verify server connectivity Test if your MCP server starts correctly: ```bash # Test filesystem server directly python -m mcp_server_filesystem --root . # Test with specific arguments npx @modelcontextprotocol/server-git --repository . ``` #### 2. Check tool discovery Use the debug script to verify tools are discovered: ```python # Check if tools are found tools = await provider.list_tools() if not tools: print("No tools discovered - check server configuration") else: print(f"Discovered {len(tools)} tools") ``` #### 3. Test tool execution Execute a simple tool to verify functionality: ```python # Test tool execution try: result = await provider.execute_tool("list_files", {"path": "."}) print(f"Tool executed successfully: {result}") except Exception as e: print(f"Tool execution failed: {e}") ``` #### 4. Monitor with debug UI 1. Start your agent with debug UI enabled 1. Navigate to [`http://localhost:9001`](http://localhost:9001) 1. Trigger tool operations through your agent 1. Review the event trace for errors or performance issues ### Common troubleshooting scenarios #### Server fails to start **Symptoms:** No tools discovered, connection timeouts **Solutions:** - Verify server command and arguments are correct - Check that required dependencies are installed - Test server startup independently - Review server logs for startup errors ```bash # Test server startup python -m mcp_server_filesystem --root . --verbose # Check dependencies pip list | grep mcp ``` #### Tools not appearing in agent **Symptoms:** [`list_tools()`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/tools.py:9) returns empty list **Solutions:** - Verify server configuration in agent YAML - Check server name matches configuration - Ensure transport type is correct - Test with debug script ```python # Verify configuration config = { "servers": [ { "name": "test_server", # Check this name "transport": "stdio", # Verify transport type "command": ["python", "-m", "mcp_server_filesystem"], "args": ["--root", "."] } ] } ``` #### Authentication failures **Symptoms:** Connection refused, 401/403 errors **Solutions:** - Verify API keys and tokens are valid - Check environment variables are set - Confirm header format matches server expectations - Test authentication independently ```bash # Check environment variables echo $API_TOKEN echo $CLIENT_ID # Test API endpoint directly curl -H "Authorization: Bearer $API_TOKEN" https://api.example.com/mcp ``` #### Tool execution errors **Symptoms:** Tools discovered but execution fails **Solutions:** - Verify parameter types match tool schema - Check parameter names are correct - Review tool documentation for requirements - Test with minimal parameters first ```python # Debug tool parameters tool = tools[0] # Get first tool print(f"Required parameters: {[p.name for p in tool.parameters if p.required]}") print(f"Optional parameters: {[p.name for p in tool.parameters if not p.required]}") # Test with minimal parameters minimal_params = {p.name: "test_value" for p in tool.parameters if p.required} result = await provider.execute_tool(tool.name, minimal_params) ``` #### Performance issues **Symptoms:** Slow tool responses, timeouts **Solutions:** - Increase timeout values in configuration - Check network latency for remote servers - Monitor server resource usage - Use debug UI to identify bottlenecks ```yaml # Increase timeouts config: timeout: 120.0 # Increase from default 60 seconds servers: - name: slow_server transport: sse url: "https://slow-api.example.com/mcp" ``` ### Debug UI features The debug UI at [`http://localhost:9001`](http://localhost:9001) provides: #### Event trace viewer - Real-time display of all tool-related events - Expandable event details with full context - Filtering by event type or tool name - Export capabilities for offline analysis Access these features by navigating to the debug UI after starting your development server with [`uv run xaibo dev`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/cli/__init__.py:272). # How to create and integrate Python tools This guide shows you how to create custom Python functions and make them available as tools for your Xaibo agents. ## Create a Python tool function Create a new Python file in your project's `tools/` directory: ```python # tools/my_tools.py from datetime import datetime, timezone from xaibo.primitives.modules.tools.python_tool_provider import tool @tool def current_time(): """Gets the current time in UTC""" return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") @tool def calculate_sum(numbers: list[float]) -> float: """Calculates the sum of a list of numbers""" return sum(numbers) @tool def format_text(text: str, style: str = "uppercase") -> str: """Formats text according to the specified style Args: text: The text to format style: Format style - 'uppercase', 'lowercase', or 'title' """ if style == "uppercase": return text.upper() elif style == "lowercase": return text.lower() elif style == "title": return text.title() else: return text ``` ## Add tools to your agent configuration Configure your agent to use the Python tool provider: ```yaml # agents/my_agent.yml id: my-agent description: An agent with custom Python tools modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: llm config: model: gpt-4.1-nano - id: python-tools module: xaibo.primitives.modules.tools.PythonToolProvider config: tool_packages: [tools.my_tools] - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator config: max_thoughts: 10 system_prompt: | You are a helpful assistant with access to custom tools. Use the available tools to help users with their requests. ``` ## Test your tools Start your agent and test the tools: ```bash # Start the development server uv run xaibo dev # Test with curl curl -X POST http://127.0.0.1:9001/openai/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "my-agent", "messages": [ {"role": "user", "content": "What time is it now?"} ] }' ``` ## Add tools with complex parameters Create tools that accept structured data: ```python # tools/advanced_tools.py from typing import Dict, List, Optional from dataclasses import dataclass from xaibo.primitives.modules.tools.python_tool_provider import tool @dataclass class Person: name: str age: int email: Optional[str] = None @tool def create_person_profile(name: str, age: int, email: str = None) -> Dict: """Creates a person profile with the given information Args: name: Person's full name age: Person's age in years email: Optional email address """ profile = { "name": name, "age": age, "profile_id": f"{name.lower().replace(' ', '_')}_{age}" } if email: profile["email"] = email return profile @tool def search_database(query: str, filters: Dict = None, limit: int = 10) -> List[Dict]: """Searches a mock database with optional filters Args: query: Search query string filters: Optional dictionary of filter criteria limit: Maximum number of results to return """ # Mock implementation results = [ {"id": 1, "title": f"Result for '{query}'", "score": 0.95}, {"id": 2, "title": f"Another match for '{query}'", "score": 0.87} ] if filters: # Apply mock filtering logic results = [r for r in results if r["score"] >= filters.get("min_score", 0)] return results[:limit] ``` ## Handle errors in tools Add proper error handling to your tools: ```python # tools/robust_tools.py import requests from xaibo.primitives.modules.tools.python_tool_provider import tool @tool def fetch_weather(city: str) -> Dict: """Fetches weather information for a city Args: city: Name of the city """ try: # Mock API call (replace with real weather API) if not city or len(city.strip()) == 0: raise ValueError("City name cannot be empty") # Simulate API response weather_data = { "city": city, "temperature": "22°C", "condition": "Sunny", "humidity": "65%" } return weather_data except ValueError as e: return {"error": f"Invalid input: {str(e)}"} except Exception as e: return {"error": f"Failed to fetch weather: {str(e)}"} @tool def validate_email(email: str) -> Dict: """Validates an email address format Args: email: Email address to validate """ import re pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' is_valid = bool(re.match(pattern, email)) return { "email": email, "is_valid": is_valid, "message": "Valid email format" if is_valid else "Invalid email format" } ``` ## Use multiple tool packages Configure multiple tool packages in your agent: ```yaml # agents/multi_tool_agent.yml id: multi-tool-agent modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: llm config: model: gpt-4.1-nano - id: python-tools module: xaibo.primitives.modules.tools.PythonToolProvider config: tool_packages: - tools.my_tools - tools.advanced_tools - tools.robust_tools - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator ``` ## Add tools from external packages Use tools from installed Python packages: ```python # tools/external_tools.py import json import base64 from xaibo.primitives.modules.tools.python_tool_provider import tool @tool def encode_base64(text: str) -> str: """Encodes text to base64""" return base64.b64encode(text.encode()).decode() @tool def decode_base64(encoded: str) -> str: """Decodes base64 text""" try: return base64.b64decode(encoded.encode()).decode() except Exception as e: return f"Error decoding: {str(e)}" @tool def format_json(data: str) -> str: """Formats JSON string with proper indentation""" try: parsed = json.loads(data) return json.dumps(parsed, indent=2) except json.JSONDecodeError as e: return f"Invalid JSON: {str(e)}" ``` ## Best practices ### Tool documentation - Always include clear docstrings with parameter descriptions - Use type hints for better tool schema generation - Document expected return formats ### Error handling - Validate input parameters - Return structured error information - Don't let exceptions crash the agent ### Tool naming - Use descriptive function names - Group related tools in the same module - Avoid name conflicts between packages ### Performance - Keep tools lightweight and fast - Use async functions for I/O operations when needed - Cache expensive computations when appropriate ## Troubleshooting ### Tools not appearing - Verify the tool package is listed in `tool_packages` - Check that functions have the `@tool` decorator - Ensure the Python module can be imported ### Import errors - Make sure all dependencies are installed - Check Python path includes your tools directory - Verify module names match file names ### Tool execution errors - Add logging to your tool functions - Check the agent debug UI for detailed error messages - Test tools independently before integrating ## Debug and troubleshoot tools Use Xaibo's built-in debugging capabilities to diagnose tool discovery, import, and execution issues. ### Enable debug UI Start your development server with the debug UI enabled: ```bash # Start with debug UI (automatically enabled in dev mode) uv run xaibo dev # Access debug UI at http://localhost:9001 ``` The debug UI provides: - Real-time event tracing for all tool operations - Visual representation of tool discovery and execution - Detailed error logs and stack traces - Performance metrics for tool calls ### Use list_tools() for programmatic debugging Access tool information programmatically using the [`list_tools()`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/tools.py:9) method: ```python # debug_python_tools.py import asyncio from xaibo.primitives.modules.tools.python_tool_provider import PythonToolProvider async def debug_python_tools(): """Debug Python tool discovery and availability""" config = { "tool_packages": ["tools.my_tools", "tools.advanced_tools"] } provider = PythonToolProvider(config=config) try: # List all available tools tools = await provider.list_tools() print(f"Found {len(tools)} Python tools:") for tool in tools: print(f" - {tool.name}") print(f" Description: {tool.description}") print(f" Parameters: {len(tool.parameters)} params") for param in tool.parameters: print(f" - {param.name}: {param.type} ({'required' if param.required else 'optional'})") print() except Exception as e: print(f"Error debugging tools: {e}") import traceback traceback.print_exc() if __name__ == "__main__": asyncio.run(debug_python_tools()) ``` ### Step-by-step debugging procedure Follow these steps to diagnose Python tool issues: #### 1. Verify tool package imports Test if your tool packages can be imported correctly: ```python # Test package imports try: import tools.my_tools print("✓ tools.my_tools imported successfully") # Check for tool functions tool_functions = [obj for obj in tools.my_tools.__dict__.values() if hasattr(obj, "__xaibo_tool__")] print(f"✓ Found {len(tool_functions)} tool functions") except ImportError as e: print(f"✗ Import failed: {e}") ``` #### 2. Check tool discovery Use the debug script to verify tools are discovered: ```python # Check if tools are found tools = await provider.list_tools() if not tools: print("No tools discovered - check package configuration") else: print(f"Discovered {len(tools)} tools") for tool in tools: print(f" - {tool.name} from {tool.description}") ``` #### 3. Test tool execution Execute a simple tool to verify functionality: ```python # Test tool execution try: result = await provider.execute_tool("current_time", {}) print(f"Tool executed successfully: {result}") except Exception as e: print(f"Tool execution failed: {e}") import traceback traceback.print_exc() ``` #### 4. Monitor with debug UI 1. Start your agent with debug UI enabled 1. Navigate to [`http://localhost:9001`](http://localhost:9001) 1. Trigger tool operations through your agent 1. Review the event trace for errors or performance issues ### Common troubleshooting scenarios #### Tools not discovered **Symptoms:** [`list_tools()`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/tools.py:9) returns empty list **Solutions:** - Verify package names in `tool_packages` configuration - Check that functions have the [`@tool`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/tools/python_tool_provider.py:12) decorator - Ensure packages can be imported from Python path - Test imports manually ```python # Verify tool package configuration config = { "tool_packages": [ "tools.my_tools", # Check this path "tools.advanced_tools" # Verify this exists ] } # Test manual import try: import tools.my_tools print("Package imports successfully") except ImportError as e: print(f"Import error: {e}") ``` #### Import errors **Symptoms:** ModuleNotFoundError, ImportError exceptions **Solutions:** - Ensure all dependencies are installed - Check Python path includes your tools directory - Verify module names match file names - Add `__init__.py` files to make directories packages ```bash # Check Python path python -c "import sys; print('\n'.join(sys.path))" # Verify package structure ls -la tools/ # Should show: # __init__.py # my_tools.py # advanced_tools.py # Test import manually python -c "import tools.my_tools; print('Import successful')" ``` #### Missing @tool decorator **Symptoms:** Functions exist but not discovered as tools **Solutions:** - Add [`@tool`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/tools/python_tool_provider.py:12) decorator to functions - Import decorator from correct module - Verify decorator is applied before function definition ```python # Correct tool definition from xaibo.primitives.modules.tools.python_tool_provider import tool @tool # This decorator is required def my_function(): """Function description""" return "result" # Check if function is marked as tool print(hasattr(my_function, "__xaibo_tool__")) # Should be True ``` #### Tool execution errors **Symptoms:** Tools discovered but execution fails **Solutions:** - Check function parameter types and names - Verify return values are JSON-serializable - Add proper error handling to tool functions - Test functions independently ```python # Debug tool parameters tool = tools[0] # Get first tool print(f"Tool name: {tool.name}") print(f"Required parameters: {[p.name for p in tool.parameters if p.required]}") print(f"Optional parameters: {[p.name for p in tool.parameters if not p.required]}") # Test with correct parameters try: # Use actual parameter names from tool definition params = {"text": "hello", "style": "uppercase"} # Example result = await provider.execute_tool(tool.name, params) print(f"Success: {result}") except Exception as e: print(f"Error: {e}") ``` #### Type annotation issues **Symptoms:** Tool schema generation fails, parameter validation errors **Solutions:** - Add proper type hints to function parameters - Use supported types (str, int, float, bool, list, dict) - Provide default values for optional parameters - Include comprehensive docstrings ```python # Good type annotations @tool def process_data( text: str, # Required string parameter count: int = 1, # Optional with default options: dict = None, # Optional complex type enabled: bool = True # Optional boolean ) -> dict: # Return type annotation """Process text data with options Args: text: Input text to process count: Number of times to process (default: 1) options: Additional processing options enabled: Whether processing is enabled Returns: Dictionary with processing results """ if options is None: options = {} return { "processed_text": text * count, "options_used": options, "was_enabled": enabled } ``` ### Debug UI features The debug UI at [`http://localhost:9001`](http://localhost:9001) provides: #### Event trace viewer - Real-time display of all tool-related events - Expandable event details with full context - Filtering by event type or tool name - Export capabilities for offline analysis Access these features by navigating to the debug UI after starting your development server with [`uv run xaibo dev`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/cli/__init__.py:272). # Reference Documentation # Reference Documentation This section provides comprehensive technical documentation for Xaibo's APIs, modules, and configuration options. The reference documentation is organized by component type and provides complete specifications for all available functionality. ## Documentation Structure ### Configuration Reference - **[Agent Configuration](agent-config/)** - Complete specification for agent configuration files - **[CLI Commands](cli/)** - Command-line interface reference ### Protocol Specifications - **[Protocols Overview](protocols/)** - Introduction to Xaibo's protocol system - **[LLM Protocol](protocols/llm/)** - Language model integration protocol - **[Tools Protocol](protocols/tools/)** - Tool provider and execution protocol - **[Memory Protocol](protocols/memory/)** - Memory storage and retrieval protocols ### Module Reference - **[LLM Modules](modules/llm/)** - Language model implementations - **[Tool Modules](modules/tools/)** - Tool provider implementations - **[Memory Modules](modules/memory/)** - Memory system implementations - **[Orchestrator Modules](modules/orchestrator/)** - Agent orchestration modules ### API Reference - **[Web Server API](api/server/)** - Web server configuration and lifecycle - **[API Adapters](api/adapters/)** - OpenAI and MCP adapter specifications - **[OpenAI Responses Adapter](api/openai-responses-adapter/)** - Advanced OpenAI-compatible response management with conversation persistence ### Integration Reference - **[LiveKit Integration](integrations/livekit/)** - Complete API reference for using Xaibo agents in LiveKit voice assistants ### Troubleshooting - **[Troubleshooting Reference](troubleshooting/)** - Systematic solutions for common issues including authentication, configuration, and integration problems # Agent Configuration Reference Agent configurations define the structure and behavior of Xaibo agents through YAML files. This reference provides complete specifications for all configuration options. ## AgentConfig The root configuration object for defining an agent. **Source**: [`src/xaibo/core/config.py:56`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/config.py#L56) ### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `id` | `str` | Yes | Unique identifier for the agent | | `description` | `str` | No | Human-readable description of the agent's purpose | | `modules` | `List[ModuleConfig]` | Yes | List of module configurations that comprise the agent | | `exchange` | `List[ExchangeConfig]` | No | Explicit dependency wiring between modules | ### Methods ______________________________________________________________________ #### `load_directory(directory: str) -> Dict[str, AgentConfig]` Load all agent configurations from a directory recursively. **Parameters:** - `directory` (str, required): Path to directory containing YAML agent configurations **Returns:** - `Dict[str, AgentConfig]`: Dictionary mapping file paths to AgentConfig instances **Raises:** - `ValueError`: If any YAML files cannot be parsed as valid agent configs ______________________________________________________________________ #### `from_yaml(yaml_str: str) -> AgentConfig` Load an AgentConfig from a YAML string. **Parameters:** - `yaml_str` (str, required): YAML string containing agent configuration **Returns:** - `AgentConfig`: Parsed agent configuration ______________________________________________________________________ #### `to_yaml() -> str` Convert this AgentConfig to YAML string format. **Returns:** - `str`: YAML string representation of this config ______________________________________________________________________ ## ModuleConfig Configuration for individual modules within an agent. **Source**: [`src/xaibo/core/config.py:11`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/config.py#L11) ### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `module` | `Union[str, Type]` | Yes | Python import path to the module class | | `id` | `str` | Yes | Unique identifier for this module instance | | `scope` | `Scope` | No | Module scope (default: `Scope.Instance`) | | `provides` | `List[Union[str, Type]]` | No | List of protocols this module provides | | `uses` | `List[Union[str, Type]]` | No | List of protocols this module requires | | `config` | `Dict` | No | Module-specific configuration parameters | ### Scope Enumeration **Source**: [`src/xaibo/core/config.py:7`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/config.py#L7) | Value | Description | | --- | --- | | `Scope.Instance` | Module instance is created per agent | | `Scope.Agent` | Module instance is shared across agent instances | ### Example ```yaml modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: llm scope: instance provides: [LLMProtocol] config: model: gpt-4 temperature: 0.7 max_tokens: 2048 ``` ## ExchangeConfig Configuration for dependency injection between modules. **Source**: [`src/xaibo/core/config.py:36`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/config.py#L36) ### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `module` | `str` | No\* | ID of the module that requires the dependency | | `field_name` | `str` | No | Parameter name in the module's constructor | | `protocol` | `Union[str, Type]` | Yes | Protocol interface that defines the dependency | | `provider` | `Union[str, List[str]]` | Yes | Module ID(s) that provide the implementation | \*Required except when used with ConfigOverrides ### Special Module IDs | Module ID | Description | | --- | --- | | `__entry__` | Entry point for handling incoming messages | | `__response__` | Response handler for outgoing messages | ### Example ```yaml exchange: # Connect orchestrator to LLM - module: orchestrator protocol: LLMProtocol provider: llm # Set entry point for text messages - module: __entry__ protocol: TextMessageHandlerProtocol provider: orchestrator # Multiple providers for a single protocol - module: tool_collector protocol: ToolProviderProtocol provider: [python-tools, mcp-tools] ``` ## ConfigOverrides Runtime configuration overrides for testing and dynamic configuration. **Source**: [`src/xaibo/core/config.py:51`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/config.py#L51) ### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `instances` | `Dict[str, object]` | No | Pre-instantiated objects to inject by module ID | | `exchange` | `List[ExchangeConfig]` | No | Additional exchange configurations | ### Example ```python from xaibo.core.config import ConfigOverrides, ExchangeConfig # Override with mock LLM for testing overrides = ConfigOverrides( instances={ "llm": MockLLM(responses=["Test response"]) }, exchange=[ ExchangeConfig( protocol="LLMProtocol", provider="llm" ) ] ) ``` ## Implicit Configuration Xaibo automatically infers configuration when unambiguous: ### Automatic Response Module If no module provides `ResponseProtocol`, Xaibo adds: ```yaml modules: - module: xaibo.primitives.modules.ResponseHandler id: __response__ provides: [ResponseProtocol] ``` ### Automatic Exchange Configuration When exactly one module provides a protocol that another module requires, Xaibo creates the exchange automatically. ### Automatic Entry Points When exactly one module provides a message handler protocol, Xaibo sets it as the entry point for that message type. ## Complete Example ```yaml id: example-agent description: An example agent with LLM and tools modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: llm config: model: gpt-4 temperature: 0.7 max_tokens: 2048 timeout: 60.0 - module: xaibo.primitives.modules.tools.PythonToolProvider id: python-tools config: tool_packages: [tools.example] - module: xaibo.primitives.modules.tools.MCPToolProvider id: mcp-tools config: timeout: 30.0 servers: - name: filesystem transport: stdio command: ["python", "-m", "mcp_server_filesystem"] args: ["--root", "/workspace"] - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator config: max_thoughts: 10 system_prompt: | You are a helpful assistant with access to tools. exchange: # Explicit wiring - module: orchestrator protocol: LLMProtocol provider: llm - module: orchestrator protocol: ToolProviderProtocol provider: [python-tools, mcp-tools] - module: __entry__ protocol: TextMessageHandlerProtocol provider: orchestrator ``` ## Validation Rules 1. **Unique Module IDs**: All module IDs within an agent must be unique 1. **Valid Module Paths**: Module paths must reference importable Python classes 1. **Protocol Consistency**: Protocols referenced in exchange must be provided by some module 1. **Dependency Resolution**: All required dependencies must have providers 1. **Entry Point Requirements**: At least one entry point must be configured for message handling # CLI Commands Reference Xaibo provides command-line tools for project initialization, development, and server management. ## xaibo init Initialize a new Xaibo project with recommended structure. **Source**: [`src/xaibo/cli/__init__.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/cli/__init__.py) ### Syntax ```bash uvx xaibo init ``` ### Parameters | Parameter | Type | Required | Description | | --- | --- | --- | --- | | `project_name` | `str` | Yes | Name of the project directory to create | ### Generated Structure ```text project_name/ ├── agents/ │ └── example.yml # Example agent configuration ├── modules/ │ └── __init__.py ├── tools/ │ ├── __init__.py │ └── example.py # Example tool implementation ├── tests/ │ └── test_example.py └── .env # Environment variables ``` ### Example ```bash # Create a new project uvx xaibo init my_agent_project # Navigate to project cd my_agent_project # Project is ready for development ``` ## xaibo dev Start the development server with debug UI and API adapters. ### Syntax ```bash uv run xaibo dev [options] ``` ### Default Adapters The development server automatically includes: - **OpenAI API Adapter**: Compatible with OpenAI Chat Completions API - **Debug UI Adapter**: Web interface for visualizing agent operations - **Event Tracing**: Automatic capture of all agent interactions ### Example ```bash # Start development server with defaults uv run xaibo dev ``` ## python -m xaibo.server.web Start the production web server with configurable adapters. **Source**: [`src/xaibo/server/web.py:89`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/web.py#L89) ### Syntax ```bash python -m xaibo.server.web [options] ``` ### Options | Option | Type | Default | Description | | --- | --- | --- | --- | | `--agent-dir` | `str` | `./agents` | Directory containing agent configurations | | `--adapter` | `str` | `[]` | Python path to API adapter class (repeatable) | | `--host` | `str` | `127.0.0.1` | Host address to bind the server | | `--port` | `int` | `8000` | Port number for the server | | `--debug-ui` | `bool` | `false` | Enable debug UI and event tracing | | `--openai-api-key` | `str` | `None` | API key for OpenAI adapter authentication (optional) | | `--mcp-api-key` | `str` | `None` | API key for MCP adapter authentication (optional) | ### Available Adapters | Adapter | Description | | --- | --- | | `xaibo.server.adapters.OpenAiApiAdapter` | OpenAI Chat Completions API compatibility | | `xaibo.server.adapters.OpenAiResponsesApiAdapter` | OpenAI Responses API with conversation management | | `xaibo.server.adapters.McpApiAdapter` | Model Context Protocol (MCP) server | | `xaibo.server.adapters.UiApiAdapter` | Debug UI and GraphQL API | ### Examples ```bash # Start with OpenAI adapter only python -m xaibo.server.web \ --adapter xaibo.server.adapters.OpenAiApiAdapter # Start with multiple adapters python -m xaibo.server.web \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --adapter xaibo.server.adapters.McpApiAdapter # Start with debug UI python -m xaibo.server.web \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --debug-ui true # Start with API key authentication python -m xaibo.server.web \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --openai-api-key sk-your-secret-key-here # Start with both adapters and API keys python -m xaibo.server.web \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --adapter xaibo.server.adapters.McpApiAdapter \ --openai-api-key sk-your-openai-key \ --mcp-api-key your-mcp-secret-key # Custom configuration python -m xaibo.server.web \ --agent-dir ./production-agents \ --host 0.0.0.0 \ --port 9000 \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --adapter xaibo.server.adapters.McpApiAdapter ``` ## Common Usage Patterns ### Development Workflow ```bash # 1. Initialize project uvx xaibo init my_project cd my_project # 2. Configure environment echo "OPENAI_API_KEY=sk-..." > .env # 3. Start development server uv run xaibo dev # 4. Test with curl curl -X POST http://127.0.0.1:9001/openai/chat/completions \ -H "Content-Type: application/json" \ -d '{"model": "example", "messages": [{"role": "user", "content": "Hello"}]}' ``` ### Production Deployment ```bash # Start production server with API key authentication python -m xaibo.server.web \ --agent-dir ./production-agents \ --host 0.0.0.0 \ --port 8000 \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --adapter xaibo.server.adapters.McpApiAdapter \ --openai-api-key "${CUSTOM_OPENAI_API_KEY}" \ --mcp-api-key "${MCP_API_KEY}" ``` ### MCP Server Setup ```bash # Start as MCP server only python -m xaibo.server.web \ --adapter xaibo.server.adapters.McpApiAdapter \ --port 8000 # Start with API key authentication python -m xaibo.server.web \ --adapter xaibo.server.adapters.McpApiAdapter \ --mcp-api-key your-secret-key \ --port 8000 # Use with MCP client (no authentication) curl -X POST http://localhost:8000/mcp \ -H "Content-Type: application/json" \ -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}' # Use with MCP client (with authentication) curl -X POST http://localhost:8000/mcp \ -H "Content-Type: application/json" \ -H "Authorization: Bearer your-secret-key" \ -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}' ``` ## Environment Variables API keys can be provided via environment variables instead of command line arguments: | Environment Variable | Description | Corresponding CLI Option | | --- | --- | --- | | `CUSTOM_OPENAI_API_KEY` | API key for OpenAI adapter authentication | `--openai-api-key` | | `MCP_API_KEY` | API key for MCP adapter authentication | `--mcp-api-key` | ### Environment Variable Examples ```bash # Set environment variables export CUSTOM_OPENAI_API_KEY="sk-your-openai-key" export MCP_API_KEY="your-mcp-secret-key" # Start server (API keys automatically detected) python -m xaibo.server.web \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --adapter xaibo.server.adapters.McpApiAdapter # Command line arguments override environment variables python -m xaibo.server.web \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --openai-api-key "different-key" # Overrides CUSTOM_OPENAI_API_KEY ``` ## Authentication ### OpenAI Adapter Authentication When an API key is configured for the OpenAI adapter, all requests must include a valid `Authorization` header: ```bash # Without authentication (no API key configured) curl -X POST http://localhost:8000/openai/chat/completions \ -H "Content-Type: application/json" \ -d '{"model": "agent-id", "messages": [{"role": "user", "content": "Hello"}]}' # With authentication (API key configured) curl -X POST http://localhost:8000/openai/chat/completions \ -H "Content-Type: application/json" \ -H "Authorization: Bearer sk-your-openai-key" \ -d '{"model": "agent-id", "messages": [{"role": "user", "content": "Hello"}]}' ``` **Authentication Errors:** - `401 Unauthorized` - Missing or invalid API key - `WWW-Authenticate: Bearer` header included in error responses ### MCP Adapter Authentication When an API key is configured for the MCP adapter, all JSON-RPC requests must include a valid `Authorization` header: ```bash # Without authentication (no API key configured) curl -X POST http://localhost:8000/mcp \ -H "Content-Type: application/json" \ -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}' # With authentication (API key configured) curl -X POST http://localhost:8000/mcp \ -H "Content-Type: application/json" \ -H "Authorization: Bearer your-mcp-secret-key" \ -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}' ``` **Authentication Errors:** - JSON-RPC error response with code `-32001` for authentication failures - Error messages: "Missing Authorization header", "Invalid Authorization header format", "Invalid API key" # Troubleshooting Reference This reference provides systematic solutions for common issues encountered when using Xaibo. Each section describes specific problems, their symptoms, and step-by-step resolution procedures. ## Installation Issues ### Missing Dependencies Error **Symptoms:** - `ImportError` when importing xaibo modules - `ModuleNotFoundError` for optional dependencies - CLI commands fail with import errors **Causes:** - Missing optional dependency groups - Incomplete installation - Python version incompatibility **Resolution:** 1. Verify Python version: `python --version` (requires >=3.10) 1. Install with required dependency groups: ```bash pip install xaibo[webserver,openai,anthropic] ``` 1. For development installations: ```bash pip install -e .[webserver,openai,anthropic] ``` **Available dependency groups:** - `webserver`: FastAPI, Strawberry GraphQL, watchfiles - `openai`: OpenAI client library - `anthropic`: Anthropic client library - `google`: Google GenAI library - `bedrock`: AWS Boto3 for Bedrock - `local`: Sentence transformers, tiktoken, transformers ### UV Package Manager Issues **Symptoms:** - `xaibo init` command fails - `FileNotFoundError` for uv command - Project initialization incomplete **Resolution:** 1. Install UV package manager: ```bash curl -LsSf https://astral.sh/uv/install.sh | sh ``` 1. Verify UV installation: `uv --version` 1. Ensure UV is in PATH 1. Retry project initialization ### Web Server Dependencies Missing **Symptoms:** - `ImportError: cannot import name 'XaiboWebServer'` - `xaibo dev` or `xaibo serve` commands fail **Resolution:** 1. Install webserver dependencies: ```bash pip install xaibo[webserver] ``` 1. Verify FastAPI installation: `python -c "import fastapi"` ## Configuration Problems ### Invalid Agent Configuration **Symptoms:** - `ValueError: Invalid agent config` - YAML parsing errors - Agent instantiation failures **Common causes and fixes:** **Missing required fields:** ```yaml # Incorrect - missing id modules: - module: xaibo.primitives.modules.llm.OpenAILLM # Correct id: my-agent modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: llm ``` **Invalid module paths:** ```yaml # Incorrect - wrong module path modules: - module: xaibo.llm.OpenAI id: llm # Correct modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: llm ``` **Missing exchange configuration:** ```yaml # May require explicit exchange config for ambiguous protocols exchange: - module: orchestrator protocol: LLMProtocol provider: llm ``` ### Multiple Provider Conflicts **Symptoms:** - `ValueError: Multiple providers found for protocol` - Agent configuration validation errors **Resolution:** 1. Identify conflicting providers in error message 1. Add explicit exchange configuration: ```yaml exchange: - module: target_module_id protocol: ConflictingProtocol provider: preferred_provider_id ``` 1. Remove unused provider modules if not needed ### Environment Variable Issues **Symptoms:** - Authentication errors with LLM providers - `401 Unauthorized` responses - Missing API key errors Server Authentication vs LLM Provider Authentication This section covers authentication with LLM providers (OpenAI, Anthropic, etc.). For securing your Xaibo server with API keys, see the [authentication guide](../../how-to/authentication/). **Resolution:** 1. Create `.env` file in project root: ```bash OPENAI_API_KEY=your_key_here ANTHROPIC_API_KEY=your_key_here GOOGLE_API_KEY=your_key_here AWS_ACCESS_KEY_ID=your_key_here AWS_SECRET_ACCESS_KEY=your_key_here ``` 1. Verify environment loading in code: ```python from dotenv import load_dotenv load_dotenv() ``` ## Runtime Errors ### Module Import Failures **Symptoms:** - `ImportError` during agent instantiation - `AttributeError: module has no attribute` - Module class not found errors **Resolution:** 1. Verify module path in configuration: ```python # Test module import manually from xaibo.primitives.modules.llm import OpenAILLM ``` 1. Check for typos in module names 1. Ensure all required dependencies are installed 1. Verify Python path includes project directory ### Protocol Implementation Errors **Symptoms:** - `AttributeError: Entry module does not implement TextMessageHandlerProtocol` - Missing protocol method errors - Protocol mismatch exceptions **Resolution:** 1. Verify module implements required protocols: ```python # Check module provides correct protocols module.provides = ["TextMessageHandlerProtocol"] ``` 1. Ensure protocol methods are implemented: ```python async def handle_text(self, text: str): # Implementation required ``` 1. Check exchange configuration maps protocols correctly ### Tool Integration Failures **Symptoms:** - Tool discovery errors - Function call execution failures - MCP server connection issues **Python tool errors:** 1. Verify tool package structure: ```python # tools/__init__.py must exist # tools/example.py with @tool decorator from xaibo.primitives.modules.tools.python_tool_provider import tool @tool def my_function(): """Function description""" return "result" ``` 1. Check tool package configuration: ```yaml config: tool_packages: [tools.example] # Correct import path ``` **MCP server errors:** 1. Verify server process is running 1. Check connection configuration: ```yaml config: servers: - name: my-server command: ["python", "server.py"] args: [] ``` 1. Test server connectivity manually ## Integration Problems ### LLM Provider Issues **OpenAI Integration:** - Verify API key format: `sk-...` - Check model availability - Monitor rate limits and quotas **Anthropic Integration:** - Verify API key format: `sk-ant-...` - Check model names - Handle content filtering responses **Google Integration:** - Verify API key configuration - Check model availability - Handle safety filter responses **AWS Bedrock Integration:** - Configure AWS credentials properly - Verify region availability - Check model access permissions ### MCP Server Integration **Connection Issues:** - Verify server executable permissions - Check server startup logs - Test server independently - Validate JSON-RPC communication **Protocol Compatibility:** - Ensure MCP protocol version compatibility - Verify tool schema definitions - Check resource URI formats # API Adapters Reference API adapters provide protocol-specific interfaces for interacting with Xaibo agents. They translate external API requests into Xaibo agent calls and format responses according to the target protocol specifications. Authentication Setup Guide For step-by-step authentication setup instructions, see the [authentication how-to guide](../../../how-to/authentication/). ## Available Adapters - **[OpenAiApiAdapter](#openaiapiadapter)** - OpenAI Chat Completions API compatibility - **[OpenAiResponsesApiAdapter](../openai-responses-adapter/)** - OpenAI Responses API with conversation management - **[McpApiAdapter](#mcpapiadapter)** - Model Context Protocol (MCP) server functionality - **[UiApiAdapter](#uiapiadapter)** - GraphQL API for web interface ## OpenAiApiAdapter Provides OpenAI Chat Completions API compatibility for Xaibo agents. **Source**: [`src/xaibo/server/adapters/openai.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai.py) **Class Path**: `xaibo.server.adapters.OpenAiApiAdapter` ### Constructor ```python OpenAiApiAdapter(xaibo: Xaibo, streaming_timeout=10, api_key: Optional[str] = None) ``` #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `xaibo` | `Xaibo` | Xaibo instance containing registered agents | | `streaming_timeout` | `int` | Timeout in seconds for streaming responses (default: 10) | | `api_key` | `Optional[str]` | API key for authentication. If provided, all requests must include valid Authorization header. Falls back to `OPENAI_API_KEY` environment variable if not specified. | ### API Endpoints #### GET `/openai/models` Returns list of available agents as OpenAI-compatible models. **Authentication:** Required if API key is configured. **Request Headers:** ```http Authorization: Bearer sk-your-api-key # Required if authentication enabled ``` **Response Format:** ```json { "object": "list", "data": [ { "id": "my-agent", "object": "model", "created": 0, "owned_by": "organization-owner" } ] } ``` #### POST `/openai/chat/completions` OpenAI-compatible chat completions endpoint. **Authentication:** Required if API key is configured. **Request Headers:** ```http Content-Type: application/json Authorization: Bearer sk-your-api-key # Required if authentication enabled ``` **Request Format:** ```json { "model": "agent-id", "messages": [ { "role": "system", "content": "You are a helpful assistant." }, { "role": "user", "content": "Hello, how are you?" } ], "temperature": 0.7, "max_tokens": 2048, "stream": false } ``` **Request Parameters:** | Parameter | Type | Required | Description | | --- | --- | --- | --- | | `model` | `str` | Yes | Xaibo agent ID to use | | `messages` | `List[dict]` | Yes | Conversation messages | | `stream` | `bool` | No | Enable streaming responses | | `temperature` | `float` | No | *Not yet implemented* - Sampling temperature (0.0-2.0) | | `max_tokens` | `int` | No | *Not yet implemented* - Maximum tokens to generate | | `top_p` | `float` | No | *Not yet implemented* - Nucleus sampling parameter | | `stop` | `List[str]` | No | *Not yet implemented* - Stop sequences | | `presence_penalty` | `float` | No | *Not yet implemented* - Presence penalty (-2.0 to 2.0) | | `frequency_penalty` | `float` | No | *Not yet implemented* - Frequency penalty (-2.0 to 2.0) | | `user` | `str` | No | *Not yet implemented* - User identifier | **Response Format (Non-Streaming):** ```json { "id": "chatcmpl-123", "object": "chat.completion", "created": 1677652288, "model": "agent-id", "choices": [ { "index": 0, "message": { "role": "assistant", "content": "Hello! I'm doing well, thank you for asking." }, "finish_reason": "stop" } ], "usage": { "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0 } } ``` **Response Format (Streaming):** ```text data: {"id":"chatcmpl-123","object":"chat.completion.chunk","created":1677652288,"model":"agent-id","choices":[{"index":0,"delta":{"content":"Hello"},"finish_reason":null}]} data: {"id":"chatcmpl-123","object":"chat.completion.chunk","created":1677652288,"model":"agent-id","choices":[{"index":0,"delta":{"content":"!"},"finish_reason":null}]} data: {"id":"chatcmpl-123","object":"chat.completion.chunk","created":1677652288,"model":"agent-id","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]} data: [DONE] ``` ### Message Format Support #### Text Messages Currently only text messages are supported: ```json { "role": "user", "content": "What is the weather like?" } ``` #### Image Messages (*Planned*) Image message support is planned for future releases: ```json { "role": "user", "content": [ { "type": "text", "text": "What's in this image?" }, { "type": "image_url", "image_url": { "url": "data:image/jpeg;base64,/9j/4AAQSkZJRgABAQAAAQ..." } } ] } ``` #### Function Calls (*Planned*) Function call support is planned for future releases: ```json { "role": "assistant", "content": null, "function_call": { "name": "get_weather", "arguments": "{\"city\": \"San Francisco\"}" } } ``` ### Error Handling #### Authentication Errors When API key authentication is enabled: **Missing Authorization Header:** ```http HTTP/1.1 401 Unauthorized WWW-Authenticate: Bearer { "detail": "Missing Authorization header" } ``` **Invalid API Key:** ```http HTTP/1.1 401 Unauthorized WWW-Authenticate: Bearer { "detail": "Invalid API key" } ``` #### Agent Not Found ```json { "detail": "model not found" } ``` **Note**: Token counting is not yet implemented, so the `usage` field in responses currently returns zeros. Full OpenAI-compatible error response format and specific error codes are planned for future releases. ### Example Usage #### cURL **Without Authentication:** ```bash curl -X POST http://localhost:8000/openai/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "my-agent", "messages": [ {"role": "user", "content": "Hello!"} ], "temperature": 0.7 }' ``` **With Authentication:** ```bash curl -X POST http://localhost:8000/openai/chat/completions \ -H "Content-Type: application/json" \ -H "Authorization: Bearer sk-your-api-key" \ -d '{ "model": "my-agent", "messages": [ {"role": "user", "content": "Hello!"} ], "temperature": 0.7 }' ``` #### Python (OpenAI SDK) **Without Authentication:** ```python import openai client = openai.OpenAI( base_url="http://localhost:8000/openai", api_key="not-needed" # No authentication required ) response = client.chat.completions.create( model="my-agent", messages=[ {"role": "user", "content": "Hello!"} ], temperature=0.7 ) print(response.choices[0].message.content) ``` **With Authentication:** ```python import openai client = openai.OpenAI( base_url="http://localhost:8000/openai", api_key="sk-your-api-key" # Use your configured API key ) response = client.chat.completions.create( model="my-agent", messages=[ {"role": "user", "content": "Hello!"} ], temperature=0.7 ) print(response.choices[0].message.content) ``` #### Streaming ```python stream = client.chat.completions.create( model="my-agent", messages=[{"role": "user", "content": "Tell me a story"}], stream=True ) for chunk in stream: if chunk.choices[0].delta.content is not None: print(chunk.choices[0].delta.content, end="") ``` ## McpApiAdapter Implements Model Context Protocol (MCP) server functionality. **Source**: [`src/xaibo/server/adapters/mcp.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/mcp.py) **Class Path**: `xaibo.server.adapters.McpApiAdapter` ### Constructor ```python McpApiAdapter(xaibo: Xaibo, api_key: Optional[str] = None) ``` #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `xaibo` | `Xaibo` | Xaibo instance containing registered agents | | `api_key` | `Optional[str]` | API key for authentication. If provided, all requests must include valid Authorization header. Falls back to `MCP_API_KEY` environment variable if not specified. | ### API Endpoints #### POST `/mcp/` Main MCP JSON-RPC 2.0 endpoint for all protocol communication. **Authentication:** Required if API key is configured. **Request Headers:** ```http Content-Type: application/json Authorization: Bearer your-api-key # Required if authentication enabled ``` **Request Format:** ```json { "jsonrpc": "2.0", "id": 1, "method": "method_name", "params": { "param1": "value1", "param2": "value2" } } ``` **Response Format:** ```json { "jsonrpc": "2.0", "id": 1, "result": { "data": "response_data" } } ``` **Error Response Format:** ```json { "jsonrpc": "2.0", "id": 1, "error": { "code": -32600, "message": "Invalid Request", "data": "Additional error information" } } ``` **Authentication Error Response:** ```json { "jsonrpc": "2.0", "id": null, "error": { "code": -32001, "message": "Missing Authorization header" } } ``` ### Supported MCP Methods #### `initialize` Establishes connection and exchanges capabilities. **Request:** ```json { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "clientInfo": { "name": "my-client", "version": "1.0.0" }, "capabilities": {} } } ``` **Response:** ```json { "jsonrpc": "2.0", "id": 1, "result": { "protocolVersion": "2024-11-05", "serverInfo": { "name": "xaibo-mcp-server", "version": "1.0.0" }, "capabilities": { "tools": {} } } } ``` #### `notifications/initialized` Confirms initialization completion. **Request:** ```json { "jsonrpc": "2.0", "method": "notifications/initialized" } ``` **Response:** No response (notification) #### `tools/list` Returns available Xaibo agents as MCP tools. **Request:** ```json { "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} } ``` **Response:** ```json { "jsonrpc": "2.0", "id": 2, "result": { "tools": [ { "name": "my-agent", "description": "Execute Xaibo agent 'my-agent'", "inputSchema": { "type": "object", "properties": { "message": { "type": "string", "description": "The text message to send to the agent" } }, "required": ["message"] } } ] } } ``` #### `tools/call` Executes a specific agent with provided arguments. **Request:** ```json { "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "my-agent", "arguments": { "message": "Hello, what can you help me with?" } } } ``` **Response:** ```json { "jsonrpc": "2.0", "id": 3, "result": { "content": [ { "type": "text", "text": "Hello! I'm a helpful assistant. I can help you with various tasks..." } ] } } ``` ### Agent Entry Points For agents with multiple entry points, tools are named as `agent_id.entry_point`: ```json { "tools": [ { "name": "multi-agent.text", "description": "Execute text handler for 'multi-agent'" }, { "name": "multi-agent.image", "description": "Execute image handler for 'multi-agent'" } ] } ``` ### Error Codes | Code | Description | Meaning | | --- | --- | --- | | `-32700` | Parse error | Invalid JSON | | `-32600` | Invalid Request | Missing required fields | | `-32601` | Method not found | Unsupported MCP method | | `-32602` | Invalid params | Missing agent or arguments | | `-32603` | Internal error | Agent execution failure | | `-32001` | Authentication error | Missing, invalid, or malformed Authorization header | ### Example Usage #### cURL **Without Authentication:** ```bash # Initialize connection curl -X POST http://localhost:8000/mcp/ \ -H "Content-Type: application/json" \ -d '{ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "clientInfo": {"name": "test-client", "version": "1.0.0"} } }' # List available tools curl -X POST http://localhost:8000/mcp/ \ -H "Content-Type: application/json" \ -d '{ "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} }' # Call an agent curl -X POST http://localhost:8000/mcp/ \ -H "Content-Type: application/json" \ -d '{ "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "my-agent", "arguments": {"message": "Hello!"} } }' ``` **With Authentication:** ```bash # Initialize connection curl -X POST http://localhost:8000/mcp/ \ -H "Content-Type: application/json" \ -H "Authorization: Bearer your-api-key" \ -d '{ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "clientInfo": {"name": "test-client", "version": "1.0.0"} } }' # List available tools curl -X POST http://localhost:8000/mcp/ \ -H "Content-Type: application/json" \ -H "Authorization: Bearer your-api-key" \ -d '{ "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} }' # Call an agent curl -X POST http://localhost:8000/mcp/ \ -H "Content-Type: application/json" \ -H "Authorization: Bearer your-api-key" \ -d '{ "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "my-agent", "arguments": {"message": "Hello!"} } }' ``` #### Python (MCP Client) **Without Authentication:** ```python import json import requests class MCPClient: def __init__(self, base_url: str, api_key: str = None): self.base_url = base_url self.session = requests.Session() self.request_id = 0 # Set up authentication if API key provided if api_key: self.session.headers.update({ "Authorization": f"Bearer {api_key}" }) def _call(self, method: str, params: dict = None): self.request_id += 1 payload = { "jsonrpc": "2.0", "id": self.request_id, "method": method, "params": params or {} } response = self.session.post( f"{self.base_url}/mcp/", json=payload, headers={"Content-Type": "application/json"} ) return response.json() def initialize(self): return self._call("initialize", { "protocolVersion": "2024-11-05", "clientInfo": {"name": "python-client", "version": "1.0.0"} }) def list_tools(self): return self._call("tools/list") def call_tool(self, name: str, arguments: dict): return self._call("tools/call", { "name": name, "arguments": arguments }) # Usage without authentication client = MCPClient("http://localhost:8000") # Initialize init_response = client.initialize() print("Initialized:", init_response) # List tools tools_response = client.list_tools() print("Available tools:", tools_response["result"]["tools"]) # Call agent result = client.call_tool("my-agent", {"message": "Hello!"}) print("Agent response:", result["result"]["content"]) ``` **With Authentication:** ```python # Usage with authentication client = MCPClient("http://localhost:8000", api_key="your-api-key") # Initialize init_response = client.initialize() print("Initialized:", init_response) # List tools tools_response = client.list_tools() print("Available tools:", tools_response["result"]["tools"]) # Call agent result = client.call_tool("my-agent", {"message": "Hello!"}) print("Agent response:", result["result"]["content"]) ``` ## UiApiAdapter Provides debug UI and GraphQL API for agent inspection and monitoring. **Source**: [`src/xaibo/server/adapters/ui.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/ui.py) **Class Path**: `xaibo.server.adapters.UiApiAdapter` ### Constructor ```python UiApiAdapter(xaibo: Xaibo) ``` #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `xaibo` | `Xaibo` | Xaibo instance for agent management | ### API Endpoints #### GET `/` Serves static UI files. **Response:** Static files for the web interface #### POST `/api/ui/graphql` GraphQL endpoint for querying agent data and execution traces. **Request Format:** ```json { "query": "query GetAgents { agents { id description modules { id module } } }" } ``` **Response Format:** ```json { "data": { "agents": [ { "id": "my-agent", "description": "Example agent", "modules": [ {"id": "llm", "module": "xaibo.primitives.modules.llm.OpenAILLM"} ] } ] } } ``` ### GraphQL Schema #### Agent Type ```graphql type Agent { id: String! } ``` **Note**: Currently only the `id` field is implemented. The `description`, `modules`, and `exchange` fields are planned for future releases. #### Module Type ```graphql type Module { id: String! module: String! config: JSON provides: [String!] uses: [String!] } ``` #### Exchange Type ```graphql type Exchange { module: String! protocol: String! provider: String! fieldName: String } ``` #### DebugTrace Type ```graphql type DebugTrace { agent_id: String! events: [Event!]! } ``` **Note**: The `debug_log` query returns `DebugTrace` instead of `Trace`. The `Trace` type is not implemented. ### Example Queries #### List All Agents ```graphql query GetAgents { list_agents { id } } ``` #### Get Agent Configuration ```graphql query GetAgentConfig($agentId: String!) { agent_config(agent_id: $agentId) { id modules { id module provides uses config } exchange { module protocol provider } } } ``` #### Get Debug Traces ```graphql query GetDebugLog($agentId: String!) { debug_log(agent_id: $agentId) { agent_id events { agent_id event_name event_type module_id module_class method_name time call_id caller_id arguments result exception } } } ``` ### Debug UI Features #### Agent Overview - List of all registered agents - Agent configuration visualization - Module dependency graphs #### Execution Traces - Real-time trace viewing - Filtering by agent and event type - Detailed event inspection #### Performance Metrics - Response time statistics - Token usage tracking - Error rate monitoring ## Adapter Development ### Creating Custom Adapters ```python from fastapi import FastAPI from xaibo import Xaibo class CustomApiAdapter: def __init__(self, xaibo: Xaibo): self.xaibo = xaibo def adapt(self, app: FastAPI): """Add routes to the FastAPI application""" @app.post("/custom/endpoint") async def custom_endpoint(request: CustomRequest): # Convert request to Xaibo format agent_id = request.agent_id message = request.message # Execute agent agent = self.xaibo.get_agent(agent_id) response = await agent.handle_text_message(message) # Convert response to custom format return CustomResponse( result=response.content, metadata=response.metadata ) ``` ### Adapter Registration ```python # Register custom adapter server = XaiboWebServer( xaibo=xaibo, adapters=["my.custom.CustomApiAdapter"], agent_dir="./agents" ) ``` ### Best Practices 1. **Error Handling**: Implement comprehensive error handling 1. **Validation**: Validate all input parameters 1. **Documentation**: Provide clear API documentation 1. **Testing**: Include thorough test coverage 1. **Performance**: Optimize for expected load patterns ## Configuration Examples ### Multi-Adapter Setup ```yaml # docker-compose.yml version: '3.8' services: xaibo-server: image: xaibo:latest ports: - "8000:8000" command: > python -m xaibo.server.web --agent-dir /app/agents --adapter xaibo.server.adapters.OpenAiApiAdapter --adapter xaibo.server.adapters.McpApiAdapter --adapter xaibo.server.adapters.UiApiAdapter --host 0.0.0.0 --port 8000 volumes: - ./agents:/app/agents environment: - OPENAI_API_KEY=${OPENAI_API_KEY} ``` ### Load Balancer Configuration ````nginx upstream xaibo_backend { server xaibo-1:8000; server xaibo-2:8000; server xaibo-3:8000; } server { listen 80; server_name api.example.com; ```text location /openai/ { proxy_pass http://xaibo_backend; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } location /mcp { proxy_pass http://xaibo_backend; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } ```` } # OpenAI Responses API Adapter Reference The OpenAI Responses API Adapter provides OpenAI-compatible response management endpoints for creating, retrieving, and managing model responses with conversation history support. Quick Start New to the OpenAI Responses API? Start with the [OpenAI Responses Quickstart Guide](../../../how-to/api/openai-responses-quickstart/) to get up and running in 2 minutes. **Source**: [`src/xaibo/server/adapters/openai_responses.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py) **Class Path**: `xaibo.server.adapters.openai_responses.OpenAiResponsesApiAdapter` ## Constructor ```python OpenAiResponsesApiAdapter(xaibo: Xaibo, streaming_timeout=10, responses_dir="./responses") ``` ### Parameters | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `xaibo` | [`Xaibo`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/xaibo.py#L23) | Required | Xaibo instance containing registered agents | | `streaming_timeout` | `int` | `10` | Timeout in seconds for streaming response chunks | | `responses_dir` | `str` | `"./responses"` | Directory path for storing response database and files | ## Database Schema The adapter uses SQLite for persistent storage with three main tables: ### responses Table | Column | Type | Description | | --- | --- | --- | | `id` | `TEXT PRIMARY KEY` | Unique response identifier (format: `resp_{uuid}`) | | `object` | `TEXT` | Object type, always `"response"` | | `created_at` | `INTEGER` | Unix timestamp of creation | | `status` | `TEXT` | Response status: `"in_progress"`, `"completed"`, `"failed"`, `"cancelled"` | | `error` | `TEXT` | JSON-encoded error details if status is `"failed"` | | `incomplete_details` | `TEXT` | JSON-encoded incomplete response details | | `instructions` | `TEXT` | System instructions for the response | | `max_output_tokens` | `INTEGER` | Maximum tokens to generate | | `model` | `TEXT` | Agent ID used for the response | | `output` | `TEXT` | JSON-encoded array of output items | | `parallel_tool_calls` | `BOOLEAN` | Whether parallel tool calls are enabled | | `previous_response_id` | `TEXT` | ID of previous response in conversation chain | | `reasoning` | `TEXT` | JSON-encoded reasoning data | | `store` | `BOOLEAN` | Whether to persist the response | | `temperature` | `REAL` | Sampling temperature parameter | | `text` | `TEXT` | JSON-encoded text content | | `tool_choice` | `TEXT` | JSON-encoded tool choice configuration | | `tools` | `TEXT` | JSON-encoded array of available tools | | `top_p` | `REAL` | Nucleus sampling parameter | | `truncation` | `TEXT` | Truncation strategy | | `usage` | `TEXT` | JSON-encoded token usage statistics | | `user_id` | `TEXT` | User identifier | | `metadata` | `TEXT` | JSON-encoded metadata object | | `background` | `BOOLEAN` | Whether response runs in background | ### input_items Table | Column | Type | Description | | --- | --- | --- | | `id` | `TEXT PRIMARY KEY` | Unique input item identifier (format: `msg_{uuid}`) | | `response_id` | `TEXT` | Foreign key to responses table | | `type` | `TEXT` | Input item type (e.g., `"message"`) | | `role` | `TEXT` | Message role: `"user"`, `"assistant"`, `"system"` | | `content` | `TEXT` | JSON-encoded content array | | `created_at` | `INTEGER` | Unix timestamp of creation | ### conversation_history Table | Column | Type | Description | | --- | --- | --- | | `id` | `TEXT PRIMARY KEY` | Unique history record identifier | | `response_id` | `TEXT` | Foreign key to responses table | | `previous_response_id` | `TEXT` | Previous response in conversation chain | | `conversation_data` | `TEXT` | JSON-encoded conversation messages | | `created_at` | `INTEGER` | Unix timestamp of creation | ## API Endpoints All endpoints are mounted under the `/openai` prefix. ### POST `/openai/responses` Creates a new model response with optional conversation history. #### Request Format ```json { "input": "Hello, how are you?", "model": "my-agent", "stream": false, "instructions": "You are a helpful assistant", "max_output_tokens": 2048, "parallel_tool_calls": true, "previous_response_id": "resp_abc123", "reasoning": null, "store": true, "temperature": 1.0, "text": null, "tool_choice": "auto", "tools": [], "top_p": 1.0, "truncation": "disabled", "user": "user-123", "metadata": {}, "background": false } ``` #### Request Parameters | Parameter | Type | Required | Description | | --- | --- | --- | --- | | `input` | `str` or `array` | Yes | Input text or array of input items | | `model` | `str` | Yes | Xaibo agent ID to execute | | `stream` | `bool` | No | Enable Server-Sent Events streaming (default: `false`) | | `instructions` | `str` | No | System instructions for the agent | | `max_output_tokens` | `int` | No | Maximum tokens to generate | | `parallel_tool_calls` | `bool` | No | Enable parallel tool execution (default: `true`) | | `previous_response_id` | `str` | No | Previous response ID for conversation continuity | | `reasoning` | `object` | No | Reasoning configuration | | `store` | `bool` | No | Whether to persist response (default: `true`) | | `temperature` | `float` | No | Sampling temperature (default: `1.0`) | | `text` | `object` | No | Text generation configuration | | `tool_choice` | `str` or `object` | No | Tool selection strategy (default: `"auto"`) | | `tools` | `array` | No | Available tools array (default: `[]`) | | `top_p` | `float` | No | Nucleus sampling parameter (default: `1.0`) | | `truncation` | `str` | No | Truncation strategy (default: `"disabled"`) | | `user` | `str` | No | User identifier | | `metadata` | `object` | No | Custom metadata (default: `{}`) | | `background` | `bool` | No | Run response in background (default: `false`) | #### Input Formats **Text Input:** ```json { "input": "What is the weather like today?" } ``` **Array Input:** ```json { "input": [ { "type": "message", "role": "user", "content": [ { "type": "input_text", "text": "What is the weather like today?" } ] } ] } ``` #### Response Format (Non-Streaming) ```json { "id": "resp_abc123", "object": "response", "created_at": 1677652288, "status": "completed", "error": null, "incomplete_details": null, "instructions": "You are a helpful assistant", "max_output_tokens": 2048, "model": "my-agent", "output": [ { "type": "message", "id": "msg_def456", "status": "completed", "role": "assistant", "content": [ { "type": "output_text", "text": "The weather is sunny and warm today!", "annotations": [] } ] } ], "parallel_tool_calls": true, "previous_response_id": null, "reasoning": null, "store": true, "temperature": 1.0, "text": null, "tool_choice": "auto", "tools": [], "top_p": 1.0, "truncation": "disabled", "usage": { "input_tokens": 0, "input_tokens_details": {"cached_tokens": 0}, "output_tokens": 0, "output_tokens_details": {"reasoning_tokens": 0}, "total_tokens": 0 }, "user": "user-123", "metadata": {}, "background": false } ``` #### Streaming Response Format When `stream: true`, the response uses Server-Sent Events with the following event types: **response.created:** ```json { "type": "response.created", "response": { "id": "resp_abc123", "object": "response", "created_at": 1677652288, "status": "in_progress", "model": "my-agent" }, "sequence_number": 1 } ``` **response.in_progress:** ```json { "type": "response.in_progress", "response": { "id": "resp_abc123", "status": "in_progress" }, "sequence_number": 2 } ``` **response.output_item.added:** ```json { "type": "response.output_item.added", "output_index": 0, "item": { "id": "msg_def456", "status": "in_progress", "type": "message", "role": "assistant", "content": [] }, "sequence_number": 3 } ``` **response.content_part.added:** ```json { "type": "response.content_part.added", "item_id": "msg_def456", "output_index": 0, "content_index": 0, "part": { "type": "output_text", "text": "", "annotations": [] }, "sequence_number": 4 } ``` **response.output_text.delta:** ```json { "type": "response.output_text.delta", "item_id": "msg_def456", "output_index": 0, "content_index": 0, "delta": "Hello", "sequence_number": 5 } ``` **response.content_part.done:** ```json { "type": "response.content_part.done", "item_id": "msg_def456", "output_index": 0, "content_index": 0, "part": { "type": "output_text", "text": "Hello! The weather is sunny today.", "annotations": [] }, "sequence_number": 6 } ``` **response.output_text.done:** ```json { "type": "response.output_text.done", "item_id": "msg_def456", "output_index": 0, "content_index": 0, "text": "Hello! The weather is sunny today.", "sequence_number": 7 } ``` **response.output_item.done:** ```json { "type": "response.output_item.done", "output_index": 0, "item": { "id": "msg_def456", "status": "completed", "type": "message", "role": "assistant", "content": [ { "type": "output_text", "text": "Hello! The weather is sunny today.", "annotations": [] } ] }, "sequence_number": 8 } ``` **response.completed:** ```json { "type": "response.completed", "response": { "id": "resp_abc123", "status": "completed", "output": [...], "usage": {...} }, "sequence_number": 9 } ``` **Stream termination:** ```text data: [DONE] ``` ### GET `/openai/responses/{response_id}` Retrieves a stored response by ID. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_id` | `str` | Unique response identifier | | `include` | `List[str]` | Optional fields to include (query parameter) | #### Response Format Returns the same format as [`POST /openai/responses`](#post-openairesponses) non-streaming response. #### Error Responses **Response Not Found (404):** ```json { "detail": "Response not found" } ``` ### DELETE `/openai/responses/{response_id}` Deletes a stored response and all associated data. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_id` | `str` | Unique response identifier | #### Response Format ```json { "id": "resp_abc123", "object": "response", "deleted": true } ``` #### Error Responses **Response Not Found (404):** ```json { "detail": "Response not found" } ``` ### POST `/openai/responses/{response_id}/cancel` Cancels a background response that is in progress. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_id` | `str` | Unique response identifier | #### Response Format Returns the updated response object with `status: "cancelled"`. #### Error Responses **Response Not Found (404):** ```json { "detail": "Response not found" } ``` **Not Background Response (400):** ```json { "detail": "Only background responses can be cancelled" } ``` ### GET `/openai/responses/{response_id}/input_items` Lists input items for a specific response with pagination support. #### Parameters | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `response_id` | `str` | Required | Unique response identifier | | `after` | `str` | `None` | Cursor for pagination (item ID) | | `before` | `str` | `None` | Cursor for pagination (item ID) | | `include` | `List[str]` | `[]` | Optional fields to include | | `limit` | `int` | `20` | Number of items to return (1-100) | | `order` | `str` | `"desc"` | Sort order: `"asc"` or `"desc"` | #### Response Format ```json { "object": "list", "data": [ { "id": "msg_abc123", "type": "message", "role": "user", "content": [ { "type": "input_text", "text": "Hello, how are you?" } ] } ], "first_id": "msg_abc123", "last_id": "msg_abc123", "has_more": false } ``` #### Error Responses **Response Not Found (404):** ```json { "detail": "Response not found" } ``` ## Methods ### [`adapt(app: FastAPI)`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L42) Registers the adapter's routes with a FastAPI application. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `app` | `FastAPI` | FastAPI application instance | ### [`_init_database()`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L45) Initializes the SQLite database and creates required tables. ### [`_get_db_connection()`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L109) Returns a new SQLite database connection. #### Returns | Type | Description | | --- | --- | | `sqlite3.Connection` | Database connection object | ### [`_store_response(response_data: Dict[str, Any])`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L114) Stores a response object in the database. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_data` | `Dict[str, Any]` | Response data dictionary | ### [`_store_input_items(response_id: str, input_data: Any)`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L159) Stores input items associated with a response. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_id` | `str` | Response identifier | | `input_data` | `Any` | Input data (string or array) | ### [`_get_stored_response(response_id: str) -> Optional[Dict[str, Any]]`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L198) Retrieves a stored response from the database. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_id` | `str` | Response identifier | #### Returns | Type | Description | | --- | --- | | `Optional[Dict[str, Any]]` | Response data dictionary or `None` if not found | ### [`_get_conversation_history(previous_response_id: str) -> Optional[SimpleConversation]`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L225) Retrieves conversation history from a previous response. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `previous_response_id` | `str` | Previous response identifier | #### Returns | Type | Description | | --- | --- | | `Optional[SimpleConversation]` | Conversation object or `None` if not found | ### [`_store_conversation_history(response_id: str, previous_response_id: str, conversation: SimpleConversation)`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L252) Stores conversation history for a response. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_id` | `str` | Current response identifier | | `previous_response_id` | `str` | Previous response identifier | | `conversation` | [`SimpleConversation`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/conversation/conversation.py#L14) | Conversation object | ### [`create_response(request: Request)`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L283) Main endpoint handler for creating responses. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `request` | `Request` | FastAPI request object | #### Returns | Type | Description | | --- | --- | | `Union[Dict[str, Any], StreamingResponse]` | Response data or streaming response | ### [`_handle_streaming_response(response_data: Dict[str, Any], last_user_message: str, conversation: SimpleConversation)`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L368) Handles streaming response generation with Server-Sent Events. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_data` | `Dict[str, Any]` | Response metadata | | `last_user_message` | `str` | User's message text | | `conversation` | [`SimpleConversation`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/conversation/conversation.py#L14) | Conversation context | #### Returns | Type | Description | | --- | --- | | `StreamingResponse` | FastAPI streaming response with SSE | ### [`_handle_non_streaming_response(response_data: Dict[str, Any], last_user_message: str, conversation: SimpleConversation)`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L610) Handles non-streaming response generation. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_data` | `Dict[str, Any]` | Response metadata | | `last_user_message` | `str` | User's message text | | `conversation` | [`SimpleConversation`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/conversation/conversation.py#L14) | Conversation context | #### Returns | Type | Description | | --- | --- | | `Dict[str, Any]` | Complete response data | ### [`get_response(response_id: str, include: List[str])`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L676) Endpoint handler for retrieving stored responses. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_id` | `str` | Response identifier | | `include` | `List[str]` | Optional fields to include | #### Returns | Type | Description | | --- | --- | | `Dict[str, Any]` | Response data | ### [`delete_response(response_id: str)`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L686) Endpoint handler for deleting responses. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_id` | `str` | Response identifier | #### Returns | Type | Description | | --- | --- | | `Dict[str, Any]` | Deletion confirmation | ### [`cancel_response(response_id: str)`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L711) Endpoint handler for cancelling background responses. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_id` | `str` | Response identifier | #### Returns | Type | Description | | --- | --- | | `Dict[str, Any]` | Updated response data | ### [`get_input_items(response_id: str, after: Optional[str], before: Optional[str], include: List[str], limit: int, order: str)`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai_responses.py#L732) Endpoint handler for listing input items with pagination. #### Parameters | Parameter | Type | Description | | --- | --- | --- | | `response_id` | `str` | Response identifier | | `after` | `Optional[str]` | Pagination cursor | | `before` | `Optional[str]` | Pagination cursor | | `include` | `List[str]` | Optional fields to include | | `limit` | `int` | Number of items to return | | `order` | `str` | Sort order | #### Returns | Type | Description | | --- | --- | | `Dict[str, Any]` | Paginated list of input items | ## Conversation Management The adapter supports stateful conversations through the `previous_response_id` parameter. When provided, the adapter: 1. Retrieves conversation history from the previous response 1. Reconstructs the [`SimpleConversation`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/conversation/conversation.py#L14) object 1. Adds the new user message to the conversation 1. Passes the full conversation context to the agent 1. Stores the updated conversation history after the response ### Conversation Data Format Conversation history is stored as JSON in the `conversation_data` field: ```json { "messages": [ { "role": "user", "content": "Hello, how are you?" }, { "role": "assistant", "content": "I'm doing well, thank you!" } ] } ``` ## Agent Integration The adapter integrates with Xaibo agents using [`ConfigOverrides`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/config.py#L45) to inject conversation history: ```python agent = self.xaibo.get_agent_with(agent_id, ConfigOverrides( instances={ '__conversation_history__': conversation, '__response__': streaming_handler # For streaming only }, exchange=[ExchangeConfig( protocol='ConversationHistoryProtocol', provider='__conversation_history__' )] )) ``` ### Agent Entry Points Agents can specify entry points using the format `agent_id/entry_point`: ```json { "model": "my-agent/text" } ``` If no entry point is specified, `__entry__` is used as the default. ## Error Handling The adapter implements comprehensive error handling for various failure scenarios: ### Agent Not Found ```json { "detail": "model not found" } ``` ### Missing Required Fields ```json { "detail": "input and model are required" } ``` ### Agent Execution Failure For streaming responses, failures are communicated via `response.failed` events: ```json { "type": "response.failed", "response": { "id": "resp_abc123", "status": "failed", "error": { "code": "server_error", "message": "Agent execution failed" } }, "sequence_number": 10 } ``` ## Usage Examples ### Basic Response Creation ```bash curl -X POST http://localhost:8000/openai/responses \ -H "Content-Type: application/json" \ -d '{ "input": "What is the capital of France?", "model": "my-agent" }' ``` ### Streaming Response ```bash curl -X POST http://localhost:8000/openai/responses \ -H "Content-Type: application/json" \ -d '{ "input": "Tell me a story", "model": "my-agent", "stream": true }' ``` ### Conversation Continuation ```bash curl -X POST http://localhost:8000/openai/responses \ -H "Content-Type: application/json" \ -d '{ "input": "What did I just ask you?", "model": "my-agent", "previous_response_id": "resp_abc123" }' ``` ### Retrieve Response ```bash curl http://localhost:8000/openai/responses/resp_abc123 ``` ### List Input Items ```bash curl "http://localhost:8000/openai/responses/resp_abc123/input_items?limit=10&order=asc" ``` ### Delete Response ```bash curl -X DELETE http://localhost:8000/openai/responses/resp_abc123 ``` ## Configuration The adapter requires minimal configuration beyond the Xaibo instance: ```python from xaibo import Xaibo from xaibo.server.adapters.openai_responses import OpenAiResponsesApiAdapter xaibo = Xaibo() adapter = OpenAiResponsesApiAdapter( xaibo=xaibo, streaming_timeout=30, # Increase timeout for slow agents responses_dir="./data/responses" # Custom storage directory ) ``` ## Performance Considerations - Database operations are synchronous and may block for large datasets - Streaming responses maintain active connections and consume memory - Conversation history grows over time and may require periodic cleanup - SQLite performance may degrade with very large response volumes ## Limitations - Token counting is not implemented (usage statistics return zeros) - Tool calling support is not yet implemented - Image and multimodal input support is not implemented - Background response cancellation is basic (no active task termination) - No built-in rate limiting or authentication # Web Server API Reference The Xaibo web server provides a FastAPI-based HTTP server with configurable adapters for different API protocols. It supports hot-reloading of agent configurations and comprehensive event tracing. **Source**: [`src/xaibo/server/web.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/web.py) ## XaiboWebServer Main web server class that hosts Xaibo agents with configurable API adapters. ### Constructor ```python XaiboWebServer( xaibo: Xaibo, adapters: list[str], agent_dir: str, host: str = "127.0.0.1", port: int = 8000, debug: bool = False, openai_api_key: Optional[str] = None, mcp_api_key: Optional[str] = None ) ``` #### Parameters | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `xaibo` | `Xaibo` | Required | Xaibo instance for agent management | | `adapters` | `list[str]` | Required | List of adapter class paths to load | | `agent_dir` | `str` | Required | Directory containing agent configuration files | | `host` | `str` | `"127.0.0.1"` | Host address to bind the server | | `port` | `int` | `8000` | Port number for the server | | `debug` | `bool` | `False` | Enable debug mode with UI and event tracing | | `openai_api_key` | `Optional[str]` | `None` | API key for OpenAI adapter authentication | | `mcp_api_key` | `Optional[str]` | `None` | API key for MCP adapter authentication | #### Example ```python from xaibo import Xaibo from xaibo.server.web import XaiboWebServer # Initialize Xaibo xaibo = Xaibo() # Create server with multiple adapters and API keys server = XaiboWebServer( xaibo=xaibo, adapters=[ "xaibo.server.adapters.OpenAiApiAdapter", "xaibo.server.adapters.McpApiAdapter" ], agent_dir="./agents", host="0.0.0.0", port=9000, debug=True, openai_api_key="sk-your-openai-key", mcp_api_key="your-mcp-secret-key" ) ``` ### Methods #### `start() -> None` Start the web server using uvicorn. ```python server.start() ``` **Features:** - Starts FastAPI application with uvicorn - Enables hot-reloading of agent configurations - Configures CORS middleware for cross-origin requests - Sets up event tracing if debug mode is enabled ### Configuration File Watching The server automatically watches the agent directory for changes and reloads configurations: #### Supported File Types - `.yml` files - `.yaml` files #### Watch Behavior - **Added Files**: Automatically registers new agents - **Modified Files**: Reloads and re-registers changed agents - **Deleted Files**: Unregisters removed agents - **Subdirectories**: Recursively watches all subdirectories #### Example Directory Structure ```text agents/ ├── production/ │ ├── customer_service.yml │ └── data_analysis.yml ├── development/ │ ├── test_agent.yml │ └── experimental.yml └── shared/ └── common_tools.yml ``` ### Debug Mode Features When `debug=True`, the server enables additional features: #### Event Tracing - Captures all agent interactions - Stores traces in `./debug` directory - Provides detailed execution logs #### Debug UI - Adds UI adapter automatically - Provides web interface for agent inspection - Visualizes agent execution flows #### Event Listener ```python from xaibo.server.adapters.ui import UIDebugTraceEventListener from pathlib import Path # Automatically added in debug mode event_listener = UIDebugTraceEventListener(Path("./debug")) xaibo.register_event_listener("", event_listener.handle_event) ``` ### CORS Configuration The server includes permissive CORS settings for development: ```python app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) ``` **Production Note**: Configure more restrictive CORS settings for production deployments. ### Lifecycle Management The server uses FastAPI's lifespan events for proper resource management: ```python @asynccontextmanager async def lifespan(app: FastAPI): # Startup: Start configuration file watcher watcher_task = asyncio.create_task(watch_config_files()) yield # Shutdown: Cancel watcher and cleanup watcher_task.cancel() try: await watcher_task except asyncio.CancelledError: pass ``` ## Command Line Interface The server can be started directly from the command line: ```bash python -m xaibo.server.web [options] ``` ### Command Line Arguments | Argument | Type | Default | Description | | --- | --- | --- | --- | | `--agent-dir` | `str` | `"./agents"` | Directory containing agent configurations | | `--adapter` | `str` | `[]` | Adapter class path (repeatable) | | `--host` | `str` | `"127.0.0.1"` | Host address to bind | | `--port` | `int` | `8000` | Port number | | `--debug-ui` | `bool` | `False` | Enable writing debug traces and start web ui | | `--openai-api-key` | `str` | `None` | API key for OpenAI adapter authentication | | `--mcp-api-key` | `str` | `None` | API key for MCP adapter authentication | ### Examples #### Basic Server ```bash python -m xaibo.server.web \ --agent-dir ./my-agents \ --adapter xaibo.server.adapters.OpenAiApiAdapter ``` #### Multi-Adapter Server ```bash python -m xaibo.server.web \ --agent-dir ./agents \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --adapter xaibo.server.adapters.McpApiAdapter \ --host 0.0.0.0 \ --port 9000 ``` #### Debug Server ```bash python -m xaibo.server.web \ --agent-dir ./agents \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --debug-ui true ``` ## Adapter Integration ### Adapter Loading Adapters are loaded dynamically using the `get_class_by_path` utility: ```python def get_class_by_path(path: str) -> Type: """Load a class from its import path""" parts = path.split('.') pkg = '.'.join(parts[:-1]) cls = parts[-1] package = importlib.import_module(pkg) clazz = getattr(package, cls) return clazz ``` ### Adapter Instantiation Each adapter is instantiated with the Xaibo instance and appropriate API keys: ```python for adapter in adapters: clazz = get_class_by_path(adapter) # Pass API keys to appropriate adapters based on class name class_name = clazz.__name__ if class_name == "OpenAiApiAdapter": instance = clazz(self.xaibo, api_key=self.openai_api_key) elif class_name == "McpApiAdapter": instance = clazz(self.xaibo, api_key=self.mcp_api_key) else: instance = clazz(self.xaibo) instance.adapt(self.app) ``` ### Available Adapters | Adapter | Description | Path | Endpoints | | --- | --- | --- | --- | | OpenAI API | OpenAI Chat Completions compatibility | `xaibo.server.adapters.OpenAiApiAdapter` | `/openai/models`, `/openai/chat/completions` | | OpenAI Responses API | OpenAI Responses API with conversation management | `xaibo.server.adapters.OpenAiResponsesApiAdapter` | `/openai/responses` | | MCP API | Model Context Protocol server | `xaibo.server.adapters.McpApiAdapter` | `/mcp/` | | UI API | Debug UI and GraphQL API | `xaibo.server.adapters.UiApiAdapter` | `/api/ui/graphql`, `/` (static files) | ## Error Handling ### Configuration Errors ```python # Invalid agent configuration ValueError: "Agent configuration validation error" # Adapter loading error ImportError: "Failed to load adapter class" ``` ### Runtime Errors ```python # Port already in use OSError: "Address already in use" # Permission denied PermissionError: "Permission denied accessing agent directory" ``` ### Agent Registration Errors ```python # Duplicate agent ID ValueError: "Agent ID already registered" # Invalid agent configuration ValidationError: "Agent configuration validation failed" ``` ## Performance Considerations ### File Watching - Uses `watchfiles.awatch` for efficient file system monitoring - Monitors agent directory for configuration changes - Handles large directory structures efficiently ### Agent Loading - Lazy loading of agent configurations - Incremental updates for changed files only - Sequential loading and registration of configurations ### Memory Management - Automatic cleanup of unregistered agents - Efficient event listener management - Resource cleanup on server shutdown ## Security Considerations ### File System Access - Restricts agent loading to specified directory - Validates file paths to prevent directory traversal - Sandboxes agent execution environments ### Network Security - Configurable host binding - CORS policy configuration - Request validation and sanitization ### Agent Isolation - Isolated agent execution contexts - Resource limits per agent - Error containment between agents ## Authentication The web server supports optional API key authentication for OpenAI and MCP adapters. ### Environment Variables API keys can be provided via environment variables: | Environment Variable | Description | Adapter | | --- | --- | --- | | `CUSTOM_OPENAI_API_KEY` | API key for OpenAI adapter authentication | [`OpenAiApiAdapter`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai.py) | | `MCP_API_KEY` | API key for MCP adapter authentication | [`McpApiAdapter`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/mcp.py) | ### OpenAI Adapter Authentication When `openai_api_key` is configured, the OpenAI adapter requires Bearer token authentication: #### Request Format ```http POST /openai/chat/completions Authorization: Bearer sk-your-openai-key Content-Type: application/json { "model": "agent-id", "messages": [{"role": "user", "content": "Hello"}] } ``` #### Authentication Verification **Source**: [`src/xaibo/server/adapters/openai.py:41`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/openai.py#L41) ```python async def _verify_api_key(self, credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=False))): """Verify API key for protected endpoints""" if not self.api_key: return # No API key configured, allow access if not credentials: raise HTTPException( status_code=401, detail="Missing Authorization header", headers={"WWW-Authenticate": "Bearer"} ) if credentials.credentials != self.api_key: raise HTTPException( status_code=401, detail="Invalid API key", headers={"WWW-Authenticate": "Bearer"} ) ``` #### Error Responses | Status Code | Description | Response Headers | | --- | --- | --- | | `401` | Missing Authorization header | `WWW-Authenticate: Bearer` | | `401` | Invalid API key | `WWW-Authenticate: Bearer` | ### MCP Adapter Authentication When `mcp_api_key` is configured, the MCP adapter requires Bearer token authentication: #### Request Format ```http POST /mcp Authorization: Bearer your-mcp-secret-key Content-Type: application/json { "jsonrpc": "2.0", "id": 1, "method": "tools/list" } ``` #### Authentication Verification **Source**: [`src/xaibo/server/adapters/mcp.py:52`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/server/adapters/mcp.py#L52) ```python # Verify API key if configured if self.api_key: auth_header = request.headers.get("authorization") if not auth_header: return self._create_error_response(None, -32001, "Missing Authorization header") if not auth_header.startswith("Bearer "): return self._create_error_response(None, -32001, "Invalid Authorization header format") provided_key = auth_header[7:] # Remove "Bearer " prefix if provided_key != self.api_key: return self._create_error_response(None, -32001, "Invalid API key") ``` #### Error Responses JSON-RPC 2.0 error responses with HTTP status `200`: | Error Code | Description | | --- | --- | | `-32001` | Missing Authorization header | | `-32001` | Invalid Authorization header format | | `-32001` | Invalid API key | #### Example Error Response ```json { "jsonrpc": "2.0", "id": null, "error": { "code": -32001, "message": "Invalid API key" } } ``` ### Security Best Practices #### API Key Management - Store API keys in environment variables, not in code - Use different API keys for different environments - Rotate API keys regularly - Monitor API key usage for unusual patterns #### Network Security - Use HTTPS in production environments - Configure restrictive CORS policies for production - Implement rate limiting for public endpoints - Use reverse proxies for additional security layers #### Example Production Configuration ```bash # Set environment variables export CUSTOM_OPENAI_API_KEY="sk-prod-key-$(date +%s)" export MCP_API_KEY="mcp-prod-key-$(date +%s)" # Start server with authentication python -m xaibo.server.web \ --agent-dir ./production-agents \ --adapter xaibo.server.adapters.OpenAiApiAdapter \ --adapter xaibo.server.adapters.McpApiAdapter \ --host 0.0.0.0 \ --port 8000 ``` # LiveKit Integration Reference *See also: [How to Use Xaibo Agents in LiveKit Voice Assistants](../../../how-to/integrations/livekit-voice-assistant/)* The Xaibo LiveKit integration provides classes and utilities for using Xaibo agents within LiveKit's voice assistant framework. ## Module: `xaibo.integrations.livekit` ### Classes #### [`XaiboAgentLoader`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py:14) LiveKit-Xaibo integration helper that enables direct use of Xaibo agents in LiveKit applications with YAML loading and debugging. **Constructor:** ```python XaiboAgentLoader() -> None ``` Initializes the XaiboAgentLoader with a new Xaibo instance. **Methods:** ##### [`load_agents_from_directory(directory: str) -> None`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py:33) Load all agent configurations from a directory. Recursively scans the specified directory for YAML files containing agent configurations and registers them with the Xaibo instance. **Parameters:** - `directory` (str): Path to directory containing YAML agent configurations **Raises:** - `ValueError`: If any YAML files cannot be parsed as valid agent configs - `FileNotFoundError`: If the directory does not exist ##### [`get_llm(agent_id: str) -> XaiboLLM`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py:75) Get a configured XaiboLLM instance for the specified agent. **Parameters:** - `agent_id` (str): The ID of the agent to get an LLM instance for **Returns:** - [`XaiboLLM`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/llm.py:26): A configured LLM instance ready for use with LiveKit **Raises:** - `ValueError`: If the agent ID is not found in loaded configurations ##### [`list_agents() -> List[str]`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py:97) List all available agent IDs. **Returns:** - `List[str]`: List of agent IDs that have been loaded ##### [`get_agent_info(agent_id: str) -> Dict[str, Any]`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py:106) Get agent metadata and configuration details. **Parameters:** - `agent_id` (str): The ID of the agent to get information for **Returns:** - `Dict[str, Any]`: Dictionary containing agent metadata including id, description, modules, and exchange configuration **Raises:** - `ValueError`: If the agent ID is not found ##### [`enable_debug_logging(debug_dir: str = "./debug") -> None`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py:152) Enable Xaibo's debug logging system. Enables the same debugging capabilities as the Xaibo web server, writing debug traces to the specified directory. **Parameters:** - `debug_dir` (str, optional): Directory to write debug traces to. Defaults to "./debug" **Raises:** - `ValueError`: If debug logging dependencies are not available ##### [`enable_file_watching(directory: str) -> None`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py:184) Enable automatic reloading of agent configurations when files change. Starts a background task that watches the specified directory for changes and automatically reloads agent configurations when YAML files are modified, added, or removed. **Parameters:** - `directory` (str): Directory to watch for configuration changes **Raises:** - `RuntimeError`: If file watching is already enabled for a different directory ##### [`disable_file_watching() -> None`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py:211) Disable automatic file watching if it's currently enabled. ##### [`get_xaibo_instance() -> Xaibo`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py:241) Get the underlying Xaibo instance. Provides access to the raw Xaibo instance for advanced use cases that require direct interaction with the Xaibo framework. **Returns:** - [`Xaibo`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/xaibo.py): The underlying Xaibo instance ##### [`is_debug_enabled() -> bool`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py:253) Check if debug logging is currently enabled. **Returns:** - `bool`: True if debug logging is enabled, False otherwise ##### [`get_debug_directory() -> Optional[str]`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/agent_loader.py:262) Get the current debug directory if debug logging is enabled. **Returns:** - `Optional[str]`: The debug directory path, or None if debug logging is disabled #### [`XaiboLLM`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/llm.py:26) Xaibo LLM implementation that integrates with Xaibo's agent system. Bridges LiveKit's LLM interface with Xaibo's agent-based conversational AI system, allowing Xaibo agents to be used as LLM providers in LiveKit applications. **Constructor:** ```python XaiboLLM(*, xaibo: Xaibo, agent_id: str) -> None ``` Initialize the Xaibo LLM. **Parameters:** - `xaibo` ([`Xaibo`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/xaibo.py)): The Xaibo instance to use for agent management - `agent_id` (str): The ID of the Xaibo agent to use for processing **Methods:** ##### [`chat(...) -> XaiboLLMStream`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/llm.py:57) Create a chat stream for the given context. ```python chat( *, chat_ctx: ChatContext, tools: list[FunctionTool | RawFunctionTool] | None = None, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN, tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN, extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN, ) -> XaiboLLMStream ``` **Parameters:** - `chat_ctx` (ChatContext): The chat context containing the conversation history - `tools` (list[FunctionTool | RawFunctionTool] | None, optional): Function tools available for the agent (currently not used) - `conn_options` (APIConnectOptions, optional): Connection options for the stream - `parallel_tool_calls` (NotGivenOr[bool], optional): Whether to allow parallel tool calls (currently not used) - `tool_choice` (NotGivenOr[ToolChoice], optional): Tool choice strategy (currently not used) - `extra_kwargs` (NotGivenOr\[dict[str, Any]\], optional): Additional keyword arguments (currently not used) **Returns:** - [`XaiboLLMStream`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/llm.py:170): A stream for processing the chat #### [`XaiboLLMStream`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/llm.py:170) Xaibo LLM stream implementation that handles real-time streaming responses from Xaibo agents. Provides true streaming output by using a queue-based system that streams chunks the moment they become available from the Xaibo agent. The implementation creates an agent with both conversation history and streaming response handler injected via ConfigOverrides, enabling real-time response streaming rather than simulated streaming. **Constructor:** ```python XaiboLLMStream( llm: XaiboLLM, *, chat_ctx: ChatContext, tools: list[FunctionTool | RawFunctionTool], conn_options: APIConnectOptions, xaibo: Xaibo, agent_id: str, conversation: SimpleConversation, ) -> None ``` Initialize the Xaibo LLM stream. **Parameters:** - `llm` ([`XaiboLLM`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/llm.py:26)): The parent XaiboLLM instance - `chat_ctx` (ChatContext): The chat context to process - `tools` (list[FunctionTool | RawFunctionTool]): Available function tools - `conn_options` (APIConnectOptions): Connection options - `xaibo` ([`Xaibo`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/xaibo.py)): The Xaibo instance - `agent_id` (str): The agent ID to use - `conversation` ([`SimpleConversation`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/conversation/conversation.py)): The conversation history **Methods:** ##### [`_create_streaming_response_handler(chunk_queue: Queue)`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/llm.py:257) Create a streaming response handler that puts chunks into a queue. **Parameters:** - `chunk_queue` (Queue): The queue to put streaming chunks into **Returns:** - StreamingResponse: A response handler that streams to the queue ##### [`_stream_chunks_from_queue(chunk_queue: Queue, agent_task) -> None`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/llm.py:270) Stream chunks from the queue as they become available. **Parameters:** - `chunk_queue` (Queue): The queue containing streaming text chunks - `agent_task`: The background task running the agent ##### [`_send_final_usage_chunk(total_content: str) -> None`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/llm.py:310) Send the final usage chunk with token information. **Parameters:** - `total_content` (str): The complete response content for token counting ### Functions #### [`logger`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/integrations/livekit/log.py:3) Logger instance for the LiveKit integration module. ```python logger: logging.Logger ``` Logger configured with the name "xaibo.integrations.livekit" for integration-specific logging. ## Usage Examples ### Basic Agent Loading ```python from xaibo.integrations.livekit import XaiboAgentLoader loader = XaiboAgentLoader() loader.load_agents_from_directory("./agents") llm = loader.get_llm("my-agent") ``` ### With Debug Logging ```python loader = XaiboAgentLoader() loader.load_agents_from_directory("./agents") loader.enable_debug_logging("./debug") ``` ### With File Watching ```python loader = XaiboAgentLoader() loader.load_agents_from_directory("./agents") loader.enable_file_watching("./agents") ``` ### LiveKit Integration ```python from livekit.agents import Agent from livekit.plugins import openai, silero llm = loader.get_llm("my-agent") assistant = Agent( instructions="", vad=silero.VAD.load(), stt=openai.STT(), llm=llm, tts=openai.TTS(), ) ``` ## Installation Install with LiveKit integration support: ```bash uv add xaibo[livekit] ``` ## Dependencies The LiveKit integration requires: - `livekit-agents`: Core LiveKit agents framework - `watchfiles`: For file watching functionality - Xaibo core modules and protocols Optional dependencies for speech services: - `livekit-plugins-openai`: OpenAI STT/TTS services - `livekit-plugins-silero`: Silero VAD service # LLM Modules Reference LLM modules provide implementations of the [`LLMProtocol`](../../protocols/llm/) for various language model providers. Each module handles provider-specific authentication, request formatting, and response parsing. ## OpenAILLM OpenAI language model integration supporting GPT models. **Source**: [`src/xaibo/primitives/modules/llm/openai.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/openai.py) **Module Path**: `xaibo.primitives.modules.llm.OpenAILLM` **Dependencies**: `openai` dependency group **Protocols**: Provides [`LLMProtocol`](../../protocols/llm/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `model` | `str` | `"gpt-4.1-nano"` | OpenAI model name (e.g., "gpt-4", "gpt-4.1-nano") | | `api_key` | `str` | `None` | OpenAI API key (falls back to `OPENAI_API_KEY` env var) | | `base_url` | `str` | `"https://api.openai.com/v1"` | Base URL for OpenAI API | | `timeout` | `float` | `60.0` | Request timeout in seconds | | `temperature` | `float` | `None` | Default sampling temperature | | `max_tokens` | `int` | `None` | Default maximum tokens to generate | | `top_p` | `float` | `None` | Default nucleus sampling parameter | **Note**: Additional configuration keys become default_kwargs and are passed to the OpenAI API. ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: openai-llm config: model: gpt-4 api_key: sk-... # Optional, uses OPENAI_API_KEY env var temperature: 0.7 max_tokens: 2048 timeout: 30.0 ``` ### Features - **Function Calling**: Full support for OpenAI function calling with automatic Python type to JSON Schema mapping - **Vision**: Image input support for vision-capable models - **Streaming**: Real-time response streaming - **Token Usage**: Detailed token consumption tracking ### Implementation Details #### Function Type Mapping The [`_prepare_functions`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/openai.py:114) method automatically maps Python types to JSON Schema types: - `str` → `string` - `int` → `integer` - `float` → `number` - `bool` → `boolean` - `list` → `array` - `dict` → `object` - `None` → `null` This ensures proper type validation in OpenAI function calling. ### OpenAI API Compatibility [`OpenAILLM`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/openai.py) can be used with any OpenAI API-compatible provider by configuring the `base_url` parameter: - **Cloud Providers**: SambaNova, Together AI, Groq, and other hosted services - **Local Inference**: Ollama, LM Studio, vLLM, and other local servers - **Self-Hosted**: Custom OpenAI-compatible API implementations Configure the `base_url` to point to your provider's endpoint while keeping the same OpenAI client interface and authentication patterns. ## AnthropicLLM Anthropic Claude model integration. **Source**: [`src/xaibo/primitives/modules/llm/anthropic.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/anthropic.py) **Module Path**: `xaibo.primitives.modules.llm.AnthropicLLM` **Dependencies**: `anthropic` dependency group **Protocols**: Provides [`LLMProtocol`](../../protocols/llm/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `model` | `str` | `"claude-3-opus-20240229"` | Anthropic model name | | `api_key` | `str` | `None` | Anthropic API key (falls back to `ANTHROPIC_API_KEY` env var) | | `base_url` | `str` | `None` | Custom base URL for Anthropic API | | `timeout` | `float` | `60.0` | Request timeout in seconds | | `temperature` | `float` | `None` | Default sampling temperature | | `max_tokens` | `int` | `None` | Default maximum tokens to generate | ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.llm.AnthropicLLM id: claude-llm config: model: claude-3-opus-20240229 temperature: 0.7 max_tokens: 4096 ``` ### Features - **Tool Use**: Native support for Anthropic tool use with input_schema format - **Vision**: Image analysis capabilities - **Streaming**: Real-time response streaming - **System Messages**: Dedicated system message handling (extracted separately from message flow and passed as `system` parameter) ### Implementation Details #### System Message Handling Unlike other providers, Anthropic handles system messages separately: - System messages are extracted from the conversation flow in [`_prepare_messages`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/anthropic.py:55) - Multiple system messages are combined with spaces - The combined system message is passed as the `system` parameter to the API #### Tool Use Format Anthropic uses a different tool format than OpenAI: - Tools are defined with `input_schema` instead of `parameters` - Tool calls use `tool_use` type with `input` field for arguments - Tool results use `tool_result` type with `tool_use_id` reference ## GoogleLLM Google Gemini model integration with Vertex AI support. **Source**: [`src/xaibo/primitives/modules/llm/google.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/google.py) **Module Path**: `xaibo.primitives.modules.llm.GoogleLLM` **Dependencies**: `google` dependency group **Protocols**: Provides [`LLMProtocol`](../../protocols/llm/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `model` | `str` | `"gemini-2.0-flash-001"` | Google model name | | `api_key` | `str` | `None` | Google API key (required for AI Studio mode, does not check environment variables) | | `vertexai` | `bool` | `False` | Use Vertex AI instead of AI Studio (when true, uses service account authentication) | | `project` | `str` | `None` | GCP project ID (required for Vertex AI mode) | | `location` | `str` | `"us-central1"` | Vertex AI location | | `temperature` | `float` | `None` | Default sampling temperature | | `max_tokens` | `int` | `None` | Default maximum tokens to generate (mapped to `max_output_tokens` internally) | **Note**: The `config` parameter is required for initialization. Either `api_key` (for AI Studio) or `vertexai=true` with `project` (for Vertex AI) must be provided. ### Example Configuration ```yaml # AI Studio (API key) modules: - module: xaibo.primitives.modules.llm.GoogleLLM id: gemini-llm config: model: gemini-1.5-pro api_key: AIza... temperature: 0.7 # Vertex AI (service account) modules: - module: xaibo.primitives.modules.llm.GoogleLLM id: gemini-vertex config: model: gemini-1.5-pro vertexai: true project: my-gcp-project location: us-central1 ``` ### Features - **Multimodal**: Native support for text, images, audio, and video - **Function Calling**: Google function calling with parameter mapping to FunctionDeclaration format - **Image Format Detection**: Automatic MIME type detection for images based on file extensions (supports .png, .gif, .webp, defaults to .jpeg) - **Streaming**: Real-time response streaming - **Safety Settings**: Configurable content safety filters ### Implementation Details #### Image Format Detection The [`_convert_image`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/google.py:102) method handles both data URIs and file URLs: - **Data URIs**: Extracts MIME type and base64 data automatically - **File URLs**: Detects format from extension (.png, .gif, .webp) or defaults to image/jpeg - **Vertex AI vs AI Studio**: Automatically configures client based on `vertexai` parameter #### System Message Handling System messages are extracted from the message flow and passed as the `system_instruction` parameter to the Google API, separate from the conversation contents. ## BedrockLLM AWS Bedrock model integration supporting multiple providers. **Source**: [`src/xaibo/primitives/modules/llm/bedrock.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/bedrock.py) **Module Path**: `xaibo.primitives.modules.llm.BedrockLLM` **Dependencies**: `bedrock` dependency group **Protocols**: Provides [`LLMProtocol`](../../protocols/llm/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `model` | `str` | `"anthropic.claude-v2"` | Bedrock model ID | | `region_name` | `str` | `"us-east-1"` | AWS region | | `aws_access_key_id` | `str` | `None` | AWS access key (optional) | | `aws_secret_access_key` | `str` | `None` | AWS secret key (optional) | | `timeout` | `float` | `60.0` | Request timeout in seconds | | `temperature` | `float` | `None` | Default sampling temperature | | `max_tokens` | `int` | `None` | Default maximum tokens to generate | ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.llm.BedrockLLM id: bedrock-llm config: model: anthropic.claude-v2 region_name: us-west-2 temperature: 0.7 max_tokens: 4096 ``` ### Features - **Multi-Provider**: Access to multiple model providers through Bedrock Converse API - **AWS Integration**: Native AWS authentication and billing - **Streaming**: Real-time response streaming - **Regional Deployment**: Deploy in multiple AWS regions ## LLMCombinator Combines multiple LLM instances for advanced workflows. **Source**: [`src/xaibo/primitives/modules/llm/combinator.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/combinator.py) **Module Path**: `xaibo.primitives.modules.llm.LLMCombinator` **Dependencies**: None **Protocols**: Provides [`LLMProtocol`](../../protocols/llm/), Uses [`LLMProtocol`](../../protocols/llm/) (list) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `prompts` | `List[str]` | `[]` | Specialized prompts for each LLM | ### Constructor Dependencies | Parameter | Type | Description | | --- | --- | --- | | `llms` | `List[LLMProtocol]` | List of LLM instances to combine (passed as constructor parameter) | ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: gpt4 config: model: gpt-4 - module: xaibo.primitives.modules.llm.AnthropicLLM id: claude config: model: claude-3-opus-20240229 - module: xaibo.primitives.modules.llm.LLMCombinator id: combined-llm config: prompts: - "You are a creative writing assistant." - "You are a technical analysis expert." exchange: - module: combined-llm protocol: LLMProtocol provider: [gpt4, claude] ``` ### Features - **Multi-Model**: Combine responses from multiple models - **Specialized Prompts**: Different system prompts for each model - **Response Merging**: Automatic merging of multiple responses - **Fallback**: Automatic fallback if one model fails ## MockLLM Mock LLM implementation for testing and development. **Source**: [`src/xaibo/primitives/modules/llm/mock.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/mock.py) **Module Path**: `xaibo.primitives.modules.llm.MockLLM` **Dependencies**: None **Protocols**: Provides [`LLMProtocol`](../../protocols/llm/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `responses` | `List[Dict]` | `[]` | Predefined responses in LLMResponse format | | `streaming_delay` | `int` | `0` | Delay between streaming chunks (ms) | | `streaming_chunk_size` | `int` | `3` | Characters per streaming chunk | ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.llm.MockLLM id: mock-llm config: responses: - content: "This is the first mock response." - content: "This is the second mock response." - content: "This is the third mock response." streaming_delay: 50 streaming_chunk_size: 5 ``` ### Features - **Deterministic**: Predictable responses for testing - **Cycling**: Cycles through responses list - **Streaming Simulation**: Simulates streaming with configurable delays - **No Dependencies**: No external API dependencies ## Error Handling All LLM modules handle common error scenarios: ### Authentication Errors ```python # Missing API key ValueError: "API key not provided and OPENAI_API_KEY not set" # Invalid API key Exception: "Invalid API key provided" ``` ### Rate Limiting ```python # Rate limit exceeded Exception: "Rate limit exceeded. Retry after 60 seconds" ``` ### Model Errors ```python # Model not found Exception: "Model 'invalid-model' not found" # Context length exceeded Exception: "Request exceeds maximum context length of 4096 tokens" ``` ### Network Errors ```python # Timeout Exception: "Request timed out after 60 seconds" # Connection error Exception: "Failed to connect to API endpoint" ``` ## Performance Considerations ### Request Optimization 1. **Batch Requests**: Use multiple messages in single request when possible 1. **Context Management**: Trim conversation history to stay within limits 1. **Streaming**: Use streaming for long responses to improve perceived performance 1. **Caching**: Cache responses for identical requests ### Resource Management 1. **Connection Pooling**: Reuse HTTP connections 1. **Rate Limiting**: Implement client-side rate limiting 1. **Timeout Configuration**: Set appropriate timeouts for your use case 1. **Memory Usage**: Monitor memory usage for large conversations ### Cost Optimization 1. **Model Selection**: Choose appropriate model for task complexity 1. **Token Management**: Monitor and optimize token usage 1. **Request Batching**: Combine multiple operations when possible 1. **Prompt Engineering**: Optimize prompts for efficiency # Memory Modules Reference Memory modules provide implementations of the memory protocols for storing, retrieving, and searching information. They support vector-based semantic search, text chunking, and multi-modal embeddings. ## VectorMemory General-purpose memory system using vector embeddings for semantic search. **Source**: [`src/xaibo/primitives/modules/memory/vector_memory.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/memory/vector_memory.py) **Module Path**: `xaibo.primitives.modules.memory.VectorMemory` **Dependencies**: None **Protocols**: Provides [`MemoryProtocol`](../../protocols/memory/) ### Constructor Dependencies | Parameter | Type | Description | | --- | --- | --- | | `chunker` | `ChunkingProtocol` | Text chunking implementation | | `embedder` | `EmbeddingProtocol` | Embedding generation implementation | | `vector_index` | `VectorIndexProtocol` | Vector storage and search implementation | ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `memory_file_path` | `str` | Required | Path to pickle file for storing memories | ### Example Configuration ```yaml modules: # Chunking component - module: xaibo.primitives.modules.memory.TokenChunker id: chunker config: window_size: 512 window_overlap: 50 # Embedding component - module: xaibo.primitives.modules.memory.SentenceTransformerEmbedder id: embedder config: model_name: "all-MiniLM-L6-v2" # Vector index component - module: xaibo.primitives.modules.memory.NumpyVectorIndex id: vector_index config: storage_dir: "./memory_index" # Complete memory system - module: xaibo.primitives.modules.memory.VectorMemory id: memory config: memory_file_path: "./memories.pkl" exchange: - module: memory protocol: ChunkingProtocol provider: chunker - module: memory protocol: EmbeddingProtocol provider: embedder - module: memory protocol: VectorIndexProtocol provider: vector_index ``` ### Features - **Semantic Search**: Vector-based similarity search - **Automatic Chunking**: Splits large texts into manageable chunks - **Metadata Support**: Stores arbitrary metadata with memories - **Persistence**: Saves memories to disk for persistence - **Deduplication**: Handles similar content intelligently ## TokenChunker Splits text based on token counts for optimal embedding. **Source**: [`src/xaibo/primitives/modules/memory/token_chunker.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/memory/token_chunker.py) **Module Path**: `xaibo.primitives.modules.memory.TokenChunker` **Dependencies**: `local` dependency group (for `tiktoken`) **Protocols**: Provides [`ChunkingProtocol`](../../protocols/memory/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `window_size` | `int` | `512` | Maximum number of tokens per chunk | | `window_overlap` | `int` | `50` | Number of tokens to overlap between chunks | | `encoding_name` | `str` | `"cl100k_base"` | Tiktoken encoding to use | ### Supported Encodings | Encoding | Description | Used By | | --- | --- | --- | | `cl100k_base` | GPT-4, gpt-4.1-nano | OpenAI models | | `p50k_base` | GPT-3 (davinci, curie, etc.) | Legacy OpenAI models | | `r50k_base` | GPT-3 (ada, babbage) | Legacy OpenAI models | | `gpt2` | GPT-2 | GPT-2 models | ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.memory.TokenChunker id: chunker config: window_size: 1024 window_overlap: 100 encoding_name: "cl100k_base" ``` ### Features - **Token-Aware**: Respects model token limits - **Overlap Support**: Maintains context between chunks - **Multiple Encodings**: Supports various tokenization schemes - **Efficient**: Fast tokenization using tiktoken ## SentenceTransformerEmbedder Uses Sentence Transformers for text embeddings. **Source**: [`src/xaibo/primitives/modules/memory/sentence_transformer_embedder.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/memory/sentence_transformer_embedder.py) **Module Path**: `xaibo.primitives.modules.memory.SentenceTransformerEmbedder` **Dependencies**: `local` dependency group (for `sentence-transformers`) **Protocols**: Provides [`EmbeddingProtocol`](../../protocols/memory/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `model_name` | `str` | `"all-MiniLM-L6-v2"` | Sentence Transformer model name | | `model_kwargs` | `dict` | `{}` | Additional model constructor arguments | ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.memory.SentenceTransformerEmbedder id: embedder config: model_name: "all-mpnet-base-v2" model_kwargs: cache_folder: "./model_cache" device: "cuda" ``` ### Features - **High Quality**: State-of-the-art embedding quality - **Multiple Models**: Wide selection of pre-trained models - **GPU Support**: Automatic GPU acceleration when available - **Caching**: Model caching for faster startup ## HuggingFaceEmbedder Leverages Hugging Face models for embeddings with multi-modal support. **Source**: [`src/xaibo/primitives/modules/memory/huggingface_embedder.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/memory/huggingface_embedder.py) **Module Path**: `xaibo.primitives.modules.memory.HuggingFaceEmbedder` **Dependencies**: `local` dependency group (for `transformers`) **Protocols**: Provides [`EmbeddingProtocol`](../../protocols/memory/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `model_name` | `str` | `"sentence-transformers/all-MiniLM-L6-v2"` | Hugging Face model name | | `device` | `str` | `"cuda" if available, else "cpu"` | Device to run model on | | `max_length` | `int` | `512` | Maximum sequence length | | `pooling_strategy` | `str` | `"mean"` | Token pooling strategy | | `audio_sampling_rate` | `int` | `16000` | Audio sampling rate | | `audio_max_length` | `int` | `30` | Maximum audio length in seconds | | `audio_return_tensors` | `str` | `"pt"` | Audio tensor format | ### Pooling Strategies | Strategy | Description | Use Case | | --- | --- | --- | | `mean` | Average of all token embeddings | General text embedding | | `cls` | Use [CLS] token embedding | Classification tasks | | `max` | Max pooling over token embeddings | Capturing important features | ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.memory.HuggingFaceEmbedder id: embedder config: model_name: "microsoft/DialoGPT-medium" device: "cuda" max_length: 1024 pooling_strategy: "mean" ``` ### Features - **Multi-Modal**: Supports text, image, and audio embeddings - **Flexible Models**: Use any Hugging Face transformer model - **Custom Pooling**: Multiple pooling strategies - **Audio Support**: Built-in audio processing capabilities ## OpenAIEmbedder Utilizes OpenAI's embedding models. **Source**: [`src/xaibo/primitives/modules/memory/openai_embedder.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/memory/openai_embedder.py) **Module Path**: `xaibo.primitives.modules.memory.OpenAIEmbedder` **Dependencies**: `openai` dependency group **Protocols**: Provides [`EmbeddingProtocol`](../../protocols/memory/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `model` | `str` | `"text-embedding-ada-002"` | OpenAI embedding model | | `api_key` | `str` | `None` | OpenAI API key (falls back to OPENAI_API_KEY env var) | | `base_url` | `str` | `"https://api.openai.com/v1"` | OpenAI API base URL | | `timeout` | `float` | `60.0` | Request timeout in seconds | ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.memory.OpenAIEmbedder id: embedder config: model: "text-embedding-3-large" timeout: 30.0 ``` ### Features - **High Quality**: State-of-the-art embedding quality - **Scalable**: Cloud-based, no local compute required - **Configurable Dimensions**: Adjust dimensions for performance/quality trade-off - **Rate Limiting**: Built-in rate limiting and retry logic ## NumpyVectorIndex Simple vector index using NumPy for storage and retrieval. **Source**: [`src/xaibo/primitives/modules/memory/numpy_vector_index.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/memory/numpy_vector_index.py) **Module Path**: `xaibo.primitives.modules.memory.NumpyVectorIndex` **Dependencies**: `numpy` (core dependency) **Protocols**: Provides [`VectorIndexProtocol`](../../protocols/memory/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `storage_dir` | `str` | Required | Directory for storing vector and attribute files | ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.memory.NumpyVectorIndex id: vector_index config: storage_dir: "./vector_storage" ``` ### Storage Format The index stores data in the specified directory: ```text vector_storage/ ├── vectors/ # Directory containing individual vector files │ ├── vector_0.npy # Individual vector embeddings │ ├── vector_1.npy │ └── ... └── attributes.pkl # Pickled metadata attributes ``` ### Features - **Simple Implementation**: Easy to understand and debug - **Persistent Storage**: Saves vectors to disk - **Cosine Similarity**: Uses cosine similarity for search - **Memory Efficient**: Loads vectors on demand ## MemoryToolProvider Tool provider that exposes memory functionality through tools. **Source**: [`src/xaibo/primitives/modules/memory/memory_provider.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/memory/memory_provider.py) **Module Path**: `xaibo.primitives.modules.memory.MemoryToolProvider` **Dependencies**: None **Protocols**: Provides [`ToolProviderProtocol`](../../protocols/tools/), Uses [`MemoryProtocol`](../../protocols/memory/) ### Constructor Dependencies | Parameter | Type | Description | | --- | --- | --- | | `memory_provider` | `MemoryProtocol` | Memory system to expose through tools | ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `config` | `dict` | `None` | Optional configuration dictionary | ### Available Tools | Tool Name | Description | Parameters | | --- | --- | --- | | `store_memory` | Store a new memory in the system | `text` (string), `attributes` (object, optional) | | `get_memory` | Retrieve a specific memory by ID | `memory_id` (string) | | `search_memory` | Search memories semantically using a text query | `query` (string), `k` (integer, default: 10) | | `list_memories` | List all stored memories | None | | `delete_memory` | Delete a memory by ID | `memory_id` (string) | | `update_memory` | Update an existing memory | `memory_id` (string), `text` (string), `attributes` (object, optional) | ### Example Configuration ```yaml modules: # Memory system - module: xaibo.primitives.modules.memory.VectorMemory id: memory_system config: memory_file_path: "./memories.pkl" # Memory tool provider - module: xaibo.primitives.modules.memory.MemoryToolProvider id: memory_tools exchange: - module: memory_tools protocol: MemoryProtocol provider: memory_system ``` ### Features - **Tool Integration**: Exposes memory operations as tools - **Complete API**: All memory operations available as tools - **Error Handling**: Proper error responses for tool failures - **Metadata Support**: Supports arbitrary metadata attributes ## Common Configuration Patterns ### Basic Memory Setup ```yaml modules: - module: xaibo.primitives.modules.memory.TokenChunker id: chunker config: window_size: 512 window_overlap: 50 - module: xaibo.primitives.modules.memory.SentenceTransformerEmbedder id: embedder config: model_name: "all-MiniLM-L6-v2" - module: xaibo.primitives.modules.memory.NumpyVectorIndex id: vector_index config: storage_dir: "./memory" - module: xaibo.primitives.modules.memory.VectorMemory id: memory exchange: - module: memory protocol: ChunkingProtocol provider: chunker - module: memory protocol: EmbeddingProtocol provider: embedder - module: memory protocol: VectorIndexProtocol provider: vector_index ``` ### High-Performance Setup ```yaml modules: - module: xaibo.primitives.modules.memory.TokenChunker id: chunker config: window_size: 1024 window_overlap: 100 - module: xaibo.primitives.modules.memory.OpenAIEmbedder id: embedder config: model: "text-embedding-3-large" dimensions: 1536 - module: xaibo.primitives.modules.memory.NumpyVectorIndex id: vector_index config: storage_dir: "./high_perf_memory" - module: xaibo.primitives.modules.memory.VectorMemory id: memory ``` ### Multi-Modal Memory ```yaml modules: - module: xaibo.primitives.modules.memory.TokenChunker id: chunker - module: xaibo.primitives.modules.memory.HuggingFaceEmbedder id: embedder config: model_name: "microsoft/DialoGPT-medium" device: "cuda" audio_sampling_rate: 22050 - module: xaibo.primitives.modules.memory.NumpyVectorIndex id: vector_index config: storage_dir: "./multimodal_memory" - module: xaibo.primitives.modules.memory.VectorMemory id: memory ``` ### Memory with Tool Integration ```yaml modules: # Core memory components - module: xaibo.primitives.modules.memory.TokenChunker id: chunker config: window_size: 512 window_overlap: 50 - module: xaibo.primitives.modules.memory.SentenceTransformerEmbedder id: embedder config: model_name: "all-MiniLM-L6-v2" - module: xaibo.primitives.modules.memory.NumpyVectorIndex id: vector_index config: storage_dir: "./memory_index" # Memory system - module: xaibo.primitives.modules.memory.VectorMemory id: memory_system config: memory_file_path: "./memories.pkl" # Memory tool provider - module: xaibo.primitives.modules.memory.MemoryToolProvider id: memory_tools exchange: - module: memory_system protocol: ChunkingProtocol provider: chunker - module: memory_system protocol: EmbeddingProtocol provider: embedder - module: memory_system protocol: VectorIndexProtocol provider: vector_index - module: memory_tools protocol: MemoryProtocol provider: memory_system ``` ## Performance Considerations ### Embedding Performance 1. **Model Selection**: Choose appropriate model for quality/speed trade-off 1. **Batch Processing**: Process multiple texts together 1. **GPU Acceleration**: Use GPU for local embedding models 1. **Caching**: Cache embeddings for repeated content ### Vector Index Performance 1. **Index Size**: Monitor index size and performance 1. **Search Optimization**: Use approximate search for large indices 1. **Memory Usage**: Consider memory requirements for large vector sets 1. **Persistence**: Balance persistence frequency with performance ### Memory Management 1. **Chunk Size**: Optimize chunk size for embedding model 1. **Overlap Strategy**: Balance context preservation with storage 1. **Cleanup**: Implement memory cleanup for old entries 1. **Compression**: Consider vector compression for storage # Orchestrator Modules Reference Orchestrator modules coordinate agent behavior by managing interactions between LLMs, tools, and memory systems. They implement the core agent logic and decision-making processes. ## StressingToolUser An orchestrator that processes user messages by leveraging an LLM to generate responses and potentially use tools. If tool execution fails, it increases the temperature (stress level) for subsequent LLM calls, simulating cognitive stress. **Source**: [`src/xaibo/primitives/modules/orchestrator/stressing_tool_user.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/orchestrator/stressing_tool_user.py) **Module Path**: `xaibo.primitives.modules.orchestrator.StressingToolUser` **Dependencies**: None **Protocols**: Provides [`TextMessageHandlerProtocol`](https://github.com/XpressAI/xaibo/blob/main/src/xaibo/core/protocols/message_handlers.py), Uses [`LLMProtocol`](../../protocols/llm/), [`ToolProviderProtocol`](../../protocols/tools/), [`ResponseProtocol`](../../protocols/response/), [`ConversationHistoryProtocol`](https://github.com/XpressAI/xaibo/blob/main/src/xaibo/core/protocols/conversation.py) ### Constructor Dependencies | Parameter | Type | Description | | --- | --- | --- | | `response` | `ResponseProtocol` | Protocol for sending responses back to the user | | `llm` | `LLMProtocol` | Protocol for generating text using a language model | | `tool_provider` | `ToolProviderProtocol` | Protocol for accessing and executing tools | | `history` | `ConversationHistoryProtocol` | Conversation history for context | | `config` | `dict` | Optional configuration dictionary | ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `system_prompt` | `str` | `""` | Initial system prompt for the conversation | | `max_thoughts` | `int` | `10` | Maximum number of tool usage iterations | ### Methods #### `handle_text(text: str) -> None` Processes a user text message, potentially using tools to generate a response. **Parameters:** - `text`: The user's input text message **Behavior:** 1. Initializes a conversation with the system prompt and user message 1. Retrieves available tools from the tool provider 1. Iteratively generates responses and executes tools as needed 1. Increases stress level (temperature) if tool execution fails 1. Sends the final response back to the user **Stress Management:** - Starts with temperature 0.0 - Increases temperature by 0.1 for each tool execution failure - Higher temperature simulates cognitive stress in subsequent LLM calls ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.llm.openai.OpenAIProvider id: llm config: model: gpt-4 - module: xaibo.primitives.modules.tools.python_tool_provider.PythonToolProvider id: tools config: tool_packages: [tools.weather, tools.calendar] - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator config: max_thoughts: 15 system_prompt: | You are a helpful assistant with access to various tools. Always try to use tools when they can help answer questions. Think step by step and explain your reasoning. exchange: - module: orchestrator protocol: LLMProtocol provider: llm - module: orchestrator protocol: ToolProviderProtocol provider: tools - module: __entry__ protocol: TextMessageHandlerProtocol provider: orchestrator ``` ### Behavior The StressingToolUser follows this process: 1. **Conversation Setup**: Retrieves conversation history and adds system prompt if configured 1. **User Message Processing**: Adds the user's message to the conversation 1. **Tool Discovery**: Retrieves available tools from the tool provider 1. **Iterative Processing**: Up to `max_thoughts` iterations: - Generates LLM response with current stress level (temperature) - If tools are called and iteration limit not reached: - Executes all requested tools - Handles tool failures by increasing stress level - Adds tool results to conversation - If no tools called or max iterations reached, ends processing 1. **Response**: Sends the final assistant message back to the user ### Features - **Stress Simulation**: Increases LLM temperature on tool failures - **Tool Integration**: Seamlessly executes multiple tools per iteration - **Iteration Control**: Prevents infinite loops with `max_thoughts` limit - **Error Handling**: Gracefully handles tool execution failures - **Conversation Context**: Maintains full conversation history ### Tool Execution When tools are called: - All tool calls in a single LLM response are executed - Tool results are collected and added to the conversation - Failed tool executions increase the stress level by 0.1 - Tool execution stops when max thoughts are reached ### Example Usage ```python from xaibo.primitives.modules.orchestrator.stressing_tool_user import StressingToolUser # Initialize with dependencies orchestrator = StressingToolUser( response=response_handler, llm=llm_provider, tool_provider=tool_provider, history=conversation_history, config={ 'max_thoughts': 10, 'system_prompt': 'You are a helpful assistant.' } ) # Handle a user message await orchestrator.handle_text("What's the weather like in Paris?") ``` ## Custom Orchestrators To create custom orchestrators, implement the [`TextMessageHandlerProtocol`](https://github.com/XpressAI/xaibo/blob/main/src/xaibo/core/protocols/message_handlers.py): ```python from xaibo.core.protocols.message_handlers import TextMessageHandlerProtocol class CustomOrchestrator(TextMessageHandlerProtocol): async def handle_text(self, text: str) -> None: # Custom orchestration logic pass ``` # Tool Modules Reference Tool modules provide implementations of the [`ToolProviderProtocol`](../../protocols/tools/) for different tool sources. They handle tool discovery, parameter validation, and execution across various tool types. ## PythonToolProvider Converts Python functions into tools using the `@tool` decorator. **Source**: [`src/xaibo/primitives/modules/tools/python_tool_provider.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/tools/python_tool_provider.py) **Module Path**: `xaibo.primitives.modules.tools.PythonToolProvider` **Dependencies**: `docstring_parser` (core dependency) **Protocols**: Provides [`ToolProviderProtocol`](../../protocols/tools/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `tool_packages` | `List[str]` | `[]` | List of Python package paths containing tool functions | | `tool_functions` | `List[Callable]` | `[]` | Optional list of function objects to use as tools | ### Tool Decorator The `@tool` decorator converts Python functions into Xaibo tools: ```python from xaibo.primitives.modules.tools.python_tool_provider import tool @tool def current_time(): """Returns the current time in UTC""" from datetime import datetime, timezone return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") @tool def get_weather(city: str, units: str = "celsius") -> dict: """Get weather information for a city Args: city: Name of the city units: Temperature units (celsius, fahrenheit, kelvin) Returns: Weather information dictionary """ # Implementation here return {"temperature": 22, "conditions": "sunny"} ``` ### Parameter Type Mapping | Python Type | Tool Parameter Type | Description | | --- | --- | --- | | `str` | `string` | Text values | | `int` | `integer` | Integer numbers | | `float` | `number` | Floating point numbers | | `bool` | `boolean` | Boolean values | | `dict` | `object` | JSON objects | | `list` | `array` | JSON arrays | | `Union[str, None]` | `string` (optional) | Optional string | | `Optional[int]` | `integer` (optional) | Optional integer | ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.tools.PythonToolProvider id: python-tools config: tool_packages: - tools.weather - tools.calendar - tools.filesystem ``` ### Tool Package Structure ```text tools/ ├── __init__.py ├── weather.py # Weather-related tools ├── calendar.py # Calendar tools └── filesystem.py # File operations ``` Example tool package (`tools/weather.py`): ```python from xaibo.primitives.modules.tools.python_tool_provider import tool import requests @tool def get_current_weather(city: str, units: str = "celsius") -> dict: """Get current weather for a city Args: city: Name of the city to get weather for units: Temperature units (celsius, fahrenheit, kelvin) Returns: Current weather information """ # API call implementation response = requests.get(f"https://api.weather.com/v1/current", params={"city": city, "units": units}) return response.json() @tool def get_weather_forecast(city: str, days: int = 5) -> list: """Get weather forecast for multiple days Args: city: Name of the city days: Number of days to forecast (1-10) Returns: List of daily weather forecasts """ if not 1 <= days <= 10: raise ValueError("Days must be between 1 and 10") # Implementation here return [{"date": f"2024-01-{i+1}", "temp": 20+i} for i in range(days)] ``` ### Features - **Automatic Discovery**: Scans packages for `@tool` decorated functions - **Type Inference**: Automatically infers parameter types from annotations - **Docstring Parsing**: Extracts descriptions from function docstrings - **Error Handling**: Converts Python exceptions to tool errors - **Validation**: Validates parameters before function execution ## MCPToolProvider Connects to MCP (Model Context Protocol) servers to provide their tools. **Source**: [`src/xaibo/primitives/modules/tools/mcp_tool_provider.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/tools/mcp_tool_provider.py) **Module Path**: `xaibo.primitives.modules.tools.MCPToolProvider` **Dependencies**: `aiohttp`, `websockets` (core dependencies) **Protocols**: Provides [`ToolProviderProtocol`](../../protocols/tools/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `servers` | `List[dict]` | Required | List of MCP server configurations | | `timeout` | `float` | `30.0` | Timeout for server operations in seconds | ### Server Configuration Each server in the `servers` list requires: | Field | Type | Required | Description | | --- | --- | --- | --- | | `name` | `str` | Yes | Unique identifier for the server | | `transport` | `str` | Yes | Transport type: "stdio", "sse", or "websocket" | #### STDIO Transport For local process-based MCP servers: | Field | Type | Default | Description | | --- | --- | --- | --- | | `command` | `List[str]` | `[]` | Command and arguments to start the server | | `args` | `List[str]` | `[]` | Additional arguments | | `env` | `Dict[str, str]` | `{}` | Environment variables | #### SSE Transport For HTTP Server-Sent Events based servers: | Field | Type | Default | Description | | --- | --- | --- | --- | | `url` | `str` | `""` | Server URL | | `headers` | `Dict[str, str]` | `{}` | HTTP headers for authentication | #### WebSocket Transport For WebSocket-based servers: | Field | Type | Default | Description | | --- | --- | --- | --- | | `url` | `str` | `""` | WebSocket URL | | `headers` | `Dict[str, str]` | `{}` | HTTP headers for connection | ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.tools.MCPToolProvider id: mcp-tools config: timeout: 60.0 servers: # Local filesystem server - name: filesystem transport: stdio command: ["python", "-m", "mcp_server_filesystem"] args: ["--root", "/workspace"] env: LOG_LEVEL: "INFO" # Remote web search server - name: web_search transport: sse url: "https://api.example.com/mcp" headers: Authorization: "Bearer your-api-key" Content-Type: "application/json" # WebSocket database server - name: database transport: websocket url: "ws://localhost:8080/mcp" headers: X-API-Key: "your-websocket-key" ``` ### Tool Namespacing Tools from MCP servers are namespaced with the server name: ```text filesystem.read_file filesystem.write_file web_search.search web_search.get_page database.query database.insert ``` ### Features - **Multiple Transports**: Supports stdio, SSE, and WebSocket transports - **Connection Management**: Automatic connection establishment and recovery - **Tool Caching**: Caches tool definitions for performance - **Error Handling**: Robust error handling for network issues - **Concurrent Servers**: Supports multiple servers simultaneously ## ToolCollector Aggregates tools from multiple tool providers. **Source**: [`src/xaibo/primitives/modules/tools/tool_collector.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/tools/tool_collector.py) **Module Path**: `xaibo.primitives.modules.tools.ToolCollector` **Dependencies**: None **Protocols**: Provides [`ToolProviderProtocol`](../../protocols/tools/), Uses [`ToolProviderProtocol`](../../protocols/tools/) (list) ### Constructor Dependencies | Parameter | Type | Description | | --- | --- | --- | | `providers` | `List[ToolProviderProtocol]` | List of tool providers to aggregate | ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.tools.PythonToolProvider id: python-tools config: tool_packages: [tools.weather] - module: xaibo.primitives.modules.tools.MCPToolProvider id: mcp-tools config: servers: - name: filesystem transport: stdio command: ["python", "-m", "mcp_server_filesystem"] - module: xaibo.primitives.modules.tools.ToolCollector id: all-tools exchange: - module: all-tools protocol: ToolProviderProtocol provider: [python-tools, mcp-tools] ``` ### Features - **Tool Aggregation**: Combines tools from multiple providers - **Name Conflict Resolution**: Handles duplicate tool names - **Provider Routing**: Routes tool execution to correct provider - **Unified Interface**: Presents single interface for all tools ## OneShotTools Provides LLM-based tools defined with conversation templates and parameter injection. **Source**: [`src/xaibo/primitives/modules/tools/oneshot.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/tools/oneshot.py) **Module Path**: `xaibo.primitives.modules.tools.OneShotTools` **Dependencies**: None **Protocols**: Provides [`ToolProviderProtocol`](../../protocols/tools/), Uses [`LLMProtocol`](../../protocols/llm/) ### Configuration | Parameter | Type | Description | | --- | --- | --- | | `tools` | `List[OneShotTool]` | List of one-shot tool definitions | ### Constructor Dependencies | Parameter | Type | Description | | --- | --- | --- | | `llm` | `LLMProtocol` | LLM provider for tool execution | ### Tool Definition Structure Each tool in the `tools` list requires: | Field | Type | Description | | --- | --- | --- | | `name` | `str` | Unique tool identifier | | `description` | `str` | Tool description for LLM context | | `parameters` | `List[OneShotToolParameter]` | Input parameters | | `returns` | `OneShotToolReturn` | Return type specification | | `conversation` | `List[OneShotToolConversationEntry]` | Conversation template | ### Parameter Definition | Field | Type | Description | | --- | --- | --- | | `name` | `str` | Parameter name | | `type` | `str` | Parameter type (string, integer, etc.) | | `description` | `str` | Parameter description | ### Return Type Definition | Field | Type | Description | | --- | --- | --- | | `type` | `str` | Expected return type | | `description` | `str` | Return value description | ### Conversation Entry Structure | Field | Type | Description | | --- | --- | --- | | `role` | `LLMRole` | Message role (user, assistant, system) | | `message` | `List[OneShotToolConversationMessage]` | Message content | ### Message Types | Type | Description | Fields | | --- | --- | --- | | `text` | Text content with parameter injection | `text` | | `image_url` | Image content from URL or file path | `url` | ### Parameter Injection Parameters are injected into conversation templates using the syntax `$$params.parameter_name$$`: ```yaml conversation: - role: user message: - type: text text: "Analyze this image: $$params.image_path$$" - type: image_url url: "$$params.image_path$$" ``` ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.llm.OpenAIProvider id: vision-llm config: model: "gpt-4.1" - module: xaibo.primitives.modules.tools.OneShotTools id: vision-tools config: tools: - name: extract_text_from_image description: "Extract text content from an image using OCR" parameters: - name: image_path type: string description: "Path to the image file" returns: type: string description: "Extracted text content" conversation: - role: user message: - type: text text: "Please extract all text from this image and return it as plain text:" - type: image_url url: "$$params.image_path$$" - name: analyze_chart_data description: "Extract structured data from charts and graphs" parameters: - name: chart_image type: string description: "Path to chart image file" - name: data_format type: string description: "Desired output format (json, csv, table)" returns: type: string description: "Structured data in requested format" conversation: - role: user message: - type: text text: "Analyze this chart and extract the data in $$params.data_format$$ format:" - type: image_url url: "$$params.chart_image$$" exchange: - module: vision-tools protocol: ToolProviderProtocol provider: vision-llm ``` ### Use Cases - **OCR Processing**: Extract text from images using vision-capable LLMs - **Document Analysis**: Analyze document structure and content - **Data Extraction**: Extract structured data from charts, tables, forms - **Image Classification**: Classify images into categories - **Content Moderation**: Analyze images for policy compliance - **Visual Question Answering**: Answer questions about image content ### Features - **Template-Based**: Define tools using conversation templates - **Parameter Injection**: Dynamic parameter substitution in prompts - **Multi-Modal Support**: Supports both text and image inputs - **Vision Integration**: Works with vision-capable LLMs - **File Path Handling**: Automatic conversion of file paths to base64 data URIs - **Flexible Prompting**: Full control over LLM conversation structure ## TextBasedToolCallAdapter Wraps LLM providers to enable tool calling for LLMs without native function calling support by converting tool definitions to text prompts and parsing tool calls from response content. **Source**: [`src/xaibo/primitives/modules/tools/no_function_calling_adapter.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/tools/no_function_calling_adapter.py) **Module Path**: `xaibo.primitives.modules.tools.TextBasedToolCallAdapter` **Dependencies**: None **Protocols**: Provides [`LLMProtocol`](../../protocols/llm/), Uses [`LLMProtocol`](../../protocols/llm/) ### Configuration | Parameter | Type | Default | Description | | --- | --- | --- | --- | | `config` | `dict` | `{}` | Optional configuration dictionary (currently unused) | ### Constructor Dependencies | Parameter | Type | Description | | --- | --- | --- | | `llm` | `LLMProtocol` | The underlying LLM provider to wrap | ### Tool Call Format The adapter instructs LLMs to use a specific text format for tool calls: ```text TOOL: tool_name {"parameter": "value", "other_param": 123} ``` ### Example Configuration ```yaml modules: - module: xaibo.primitives.modules.llm.OpenAIProvider id: base-llm config: model: "gpt-4.1-nano" - module: xaibo.primitives.modules.tools.TextBasedToolCallAdapter id: text-based-llm exchange: - module: text-based-llm protocol: LLMProtocol provider: base-llm ``` ### Tool Prompt Generation The adapter automatically generates tool descriptions and injects them into the conversation. The format includes: - Tool name and description - Parameter details with type and requirement information - Usage instructions with the "TOOL:" prefix format - Example usage Example generated prompt: ```text Available tools: get_weather: Get the current weather in a given location Parameters: - location (required): The city and state, e.g. San Francisco, CA - unit: The temperature unit to use (celsius or fahrenheit) To use a tool, write TOOL: followed by the tool name and JSON arguments on a single line. Whenever you say 'I will now...' , you must follow that up with the appropriate TOOL: invocation. Example: TOOL: get_weather {"location": "San Francisco, CA"} ``` ### Tool Call Parsing The adapter parses tool calls from LLM responses by: 1. Scanning response content line by line for "TOOL:" prefix 1. Extracting tool name and JSON arguments 1. Creating [`LLMFunctionCall`](../../protocols/llm/) objects with unique IDs 1. Handling malformed JSON by falling back to raw input 1. Removing tool call lines from the final response content ### Message Processing The adapter modifies input messages by: - Adding tool descriptions to existing system messages - Creating a new system message if none exists - Preserving all other message content and metadata - Removing function definitions from [`LLMOptions`](../../protocols/llm/) to avoid duplication ### Features - **Text-Based Tool Calling**: Converts function calling to text-based instructions - **Automatic Tool Prompt Injection**: Adds tool descriptions to system messages - **Tool Call Parsing**: Extracts tool calls from LLM responses using "TOOL:" prefix - **Streaming Support**: Supports both regular and streaming generation modes (note: tool call detection not supported in streaming) - **Error Handling**: Gracefully handles malformed JSON in tool arguments - **Content Cleaning**: Removes tool call lines from final response content - **LLM Integration**: Works with LLMs that don't support function calling ## Common Configuration Patterns ### Multi-Source Tool Setup ```yaml modules: # Python tools for custom functions - module: xaibo.primitives.modules.tools.PythonToolProvider id: python-tools config: tool_packages: [tools.custom, tools.utilities] # MCP tools for external services - module: xaibo.primitives.modules.tools.MCPToolProvider id: mcp-tools config: servers: - name: filesystem transport: stdio command: ["python", "-m", "mcp_server_filesystem"] - name: web transport: sse url: "https://api.example.com/mcp" # Aggregate all tools - module: xaibo.primitives.modules.tools.ToolCollector id: all-tools exchange: - module: all-tools protocol: ToolProviderProtocol provider: [python-tools, mcp-tools] ``` ### LLM-Based Vision Tools ```yaml modules: - module: xaibo.primitives.modules.llm.OpenAIProvider id: vision-llm config: model: "gpt-4.1" - module: xaibo.primitives.modules.tools.OneShotTools id: vision-tools config: tools: - name: document_ocr description: "Extract text from document images" parameters: - name: document_path type: string description: "Path to document image" returns: type: string description: "Extracted text content" conversation: - role: user message: - type: text text: "Extract all text from this document:" - type: image_url url: "$$params.document_path$$" ``` ### Development vs Production Tools ```yaml # Development configuration modules: - module: xaibo.primitives.modules.tools.PythonToolProvider id: tools config: tool_packages: [tools.development, tools.testing] # Production configuration modules: - module: xaibo.primitives.modules.tools.MCPToolProvider id: tools config: servers: - name: production_api transport: sse url: "https://prod-api.example.com/mcp" headers: Authorization: "Bearer ${PROD_API_KEY}" ``` ## Error Handling Tool modules handle various error scenarios: ### Tool Discovery Errors ```python # Package not found ToolError: "Package 'tools.nonexistent' not found" # No tools found ToolError: "No tools found in package 'tools.empty'" ``` ### Execution Errors ```python # Tool not found ToolNotFoundError: "Tool 'nonexistent_tool' not found" # Parameter validation ToolParameterError: "Required parameter 'city' not provided" # Execution failure ToolExecutionError: "Tool execution failed: Connection timeout" ``` ### MCP Server Errors ```python # Connection failure MCPConnectionError: "Failed to connect to MCP server 'filesystem'" # Protocol error MCPProtocolError: "Invalid MCP response from server" # Server timeout MCPTimeoutError: "MCP server 'web_search' timed out after 30 seconds" ``` ## Performance Considerations ### Tool Loading 1. **Lazy Loading**: Load tools on first access 1. **Caching**: Cache tool definitions and schemas 1. **Parallel Loading**: Load from multiple sources concurrently 1. **Error Recovery**: Handle individual tool failures gracefully ### Execution Optimization 1. **Connection Pooling**: Reuse connections for MCP servers 1. **Batch Operations**: Group related tool calls when possible 1. **Timeout Management**: Set appropriate timeouts for different tools 1. **Resource Limits**: Implement memory and CPU limits ### Monitoring 1. **Execution Metrics**: Track tool usage and performance 1. **Error Rates**: Monitor tool failure rates 1. **Cache Hit Rates**: Monitor cache effectiveness 1. **Resource Usage**: Track memory and CPU usage ## Security Considerations ### Python Tools 1. **Code Review**: Review all tool functions for security 1. **Input Validation**: Validate all tool parameters 1. **Sandboxing**: Consider running tools in sandboxed environments 1. **Permission Checks**: Implement appropriate permission checks ### MCP Servers 1. **Authentication**: Use proper authentication for MCP servers 1. **Network Security**: Secure network connections (TLS/SSL) 1. **Input Sanitization**: Sanitize inputs before sending to servers 1. **Rate Limiting**: Implement rate limiting for external servers ### General 1. **Principle of Least Privilege**: Grant minimal necessary permissions 1. **Audit Logging**: Log all tool executions for audit trails 1. **Error Information**: Avoid exposing sensitive information in errors 1. **Resource Limits**: Implement appropriate resource limits # Protocols Overview Xaibo's protocol system defines standardized interfaces that enable modular, testable, and extensible agent architectures. Protocols establish contracts between components without coupling them to specific implementations. ## Protocol Architecture ### Protocol Definition Protocols in Xaibo are Python Protocol classes that define abstract interfaces: ```python from typing import Protocol, runtime_checkable @runtime_checkable class ExampleProtocol(Protocol): """Protocol for example functionality""" async def example_method(self, param: str) -> str: """Example method documentation""" ... ``` ### Key Features - **Runtime Checkable**: All protocols use `@runtime_checkable` for isinstance checks - **Async by Default**: Most protocol methods are async for non-blocking operations - **Type Annotated**: Complete type hints for parameters and return values - **Documentation**: Comprehensive docstrings for all methods ## Core Protocols ### Message Handler Protocols **Source**: [`src/xaibo/core/protocols/message_handlers.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/message_handlers.py) | Protocol | Purpose | | --- | --- | | `TextMessageHandlerProtocol` | Handle text-based user input | | `ImageMessageHandlerProtocol` | Process image messages | | `AudioMessageHandlerProtocol` | Handle audio input | | `VideoMessageHandlerProtocol` | Process video messages | ### Core Service Protocols | Protocol | Purpose | Reference | | --- | --- | --- | | `LLMProtocol` | Language model integration | [Specification](llm/) | | `ToolProviderProtocol` | Tool execution and management | [Specification](tools/) | | `MemoryProtocol` | Memory storage and retrieval | [Specification](memory/) | | `ResponseProtocol` | Response formatting and delivery | [Specification](memory/) | | `ConversationProtocol` | Dialog history management | [Source](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/conversation.py) | ## Protocol Implementation ### Implementing a Protocol ```python from xaibo.core.protocols.llm import LLMProtocol from xaibo.core.models.llm import LLMMessage, LLMOptions, LLMResponse class CustomLLM: """Custom LLM implementation""" def __init__(self, config: dict): self.config = config async def generate( self, messages: List[LLMMessage], options: Optional[LLMOptions] = None ) -> LLMResponse: # Implementation here pass async def generate_stream( self, messages: List[LLMMessage], options: Optional[LLMOptions] = None ) -> AsyncIterator[str]: # Streaming implementation here yield "response chunk" ``` ### Protocol Registration Modules can declare protocol support in two ways: #### Explicit Declaration ```yaml modules: - module: my.custom.LLM id: custom-llm provides: [LLMProtocol] uses: [MemoryProtocol] ``` #### Automatic Detection Xaibo automatically detects protocols through: 1. **Inheritance**: Classes inheriting from protocol classes 1. **Provides Method**: Classes with a `provides()` class method 1. **Duck Typing**: Classes implementing protocol methods ## Protocol Composition ### Dependency Injection Protocols enable clean dependency injection: ```python class Orchestrator: def __init__(self, llm: LLMProtocol, tools: ToolProviderProtocol): self.llm = llm self.tools = tools async def process(self, message: str) -> str: # Use injected dependencies response = await self.llm.generate([{"role": "user", "content": message}]) return response.content ``` ### Exchange Configuration The exchange system wires protocol dependencies: ```yaml exchange: - module: orchestrator protocol: LLMProtocol provider: openai-llm - module: orchestrator protocol: ToolProviderProtocol provider: [python-tools, mcp-tools] # Multiple providers ``` ## Protocol Validation ### Runtime Checks ```python from xaibo.core.protocols.llm import LLMProtocol # Validate implementation if isinstance(my_llm, LLMProtocol): # Safe to use as LLM response = await my_llm.generate(messages) ``` ### Type Checking Protocols work with static type checkers: ```python def use_llm(llm: LLMProtocol) -> None: # Type checker validates protocol compliance pass ``` ## Testing with Protocols ### Mock Implementations ```python class MockLLM: def __init__(self, responses: List[str]): self.responses = responses self.call_count = 0 async def generate(self, messages, options=None): response = self.responses[self.call_count % len(self.responses)] self.call_count += 1 return LLMResponse(content=response) async def generate_stream(self, messages, options=None): response = self.responses[self.call_count % len(self.responses)] self.call_count += 1 for char in response: yield char # Use in tests mock_llm = MockLLM(["Test response"]) assert isinstance(mock_llm, LLMProtocol) # True ``` ## Protocol Extension ### Custom Protocols Define domain-specific protocols: ```python from typing import Protocol, runtime_checkable @runtime_checkable class DatabaseProtocol(Protocol): """Protocol for database operations""" async def query(self, sql: str) -> List[dict]: """Execute SQL query""" ... async def insert(self, table: str, data: dict) -> str: """Insert data and return ID""" ... ``` ### Protocol Inheritance Extend existing protocols: ```python @runtime_checkable class AdvancedLLMProtocol(LLMProtocol): """Extended LLM protocol with additional capabilities""" async def analyze_sentiment(self, text: str) -> float: """Analyze sentiment score""" ... async def summarize(self, text: str, max_length: int) -> str: """Generate summary""" ... ``` ## Best Practices ### Protocol Design 1. **Single Responsibility**: Each protocol should have a focused purpose 1. **Async Methods**: Use async for I/O operations 1. **Type Annotations**: Provide complete type hints 1. **Documentation**: Include comprehensive docstrings 1. **Error Handling**: Define expected exceptions ### Implementation Guidelines 1. **Protocol Compliance**: Implement all required methods 1. **Error Propagation**: Let protocol-defined exceptions bubble up 1. **Resource Management**: Use async context managers when appropriate 1. **Configuration**: Accept configuration through constructor 1. **Testing**: Provide mock implementations for testing ### Exchange Configuration 1. **Explicit Wiring**: Use explicit exchange configs for complex dependencies 1. **List Dependencies**: Use lists for multiple providers of same protocol 1. **Entry Points**: Configure entry points for message handling 1. **Validation**: Ensure all dependencies are satisfied ## Common Patterns ### Decorator Pattern ```python class LoggingLLM: """Add logging to any LLM implementation""" def __init__(self, llm: LLMProtocol, logger): self.llm = llm self.logger = logger async def generate(self, messages, options=None): self.logger.info(f"Generating response for {len(messages)} messages") response = await self.llm.generate(messages, options) self.logger.info(f"Generated {len(response.content)} characters") return response ``` ### Composite Pattern ````python class CompositeTool: """Combine multiple tool providers""" ```text def __init__(self, providers: List[ToolProviderProtocol]): self.providers = providers async def list_tools(self): all_tools = [] for provider in self.providers: tools = await provider.list_tools() all_tools.extend(tools) return all_tools ```` # LLM Protocol Specification The LLM Protocol defines the interface for language model integration in Xaibo agents. It provides standardized methods for text generation and streaming responses. **Source**: [`src/xaibo/core/protocols/llm.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/llm.py) ## LLMProtocol The core protocol interface for language model implementations. ```python @runtime_checkable class LLMProtocol(Protocol): """Protocol for interacting with LLM models""" async def generate( self, messages: List[LLMMessage], options: Optional[LLMOptions] = None ) -> LLMResponse: """Generate a response from the LLM""" ... async def generate_stream( self, messages: List[LLMMessage], options: Optional[LLMOptions] = None ) -> AsyncIterator[str]: """Generate a streaming response from the LLM""" yield ... ``` ### Methods ______________________________________________________________________ #### `generate(messages, options=None) -> LLMResponse` Generate a complete response from the language model. **Parameters:** - `messages` (`List[LLMMessage]`, required): Conversation history and current input - `options` (`Optional[LLMOptions]`, optional): Generation parameters and tool definitions **Returns:** - `LLMResponse`: Complete response with content, tool calls, and usage statistics **Example:** ```python messages = [ LLMMessage.system("You are a helpful assistant"), LLMMessage.user("What is the capital of France?") ] options = LLMOptions( temperature=0.7, max_tokens=100 ) response = await llm.generate(messages, options) print(response.content) # "The capital of France is Paris." ``` ______________________________________________________________________ #### `generate_stream(messages, options=None) -> AsyncIterator[str]` Generate a streaming response from the language model. **Parameters:** - `messages` (`List[LLMMessage]`, required): Conversation history and current input - `options` (`Optional[LLMOptions]`, optional): Generation parameters and tool definitions **Returns:** - `AsyncIterator[str]`: Stream of text chunks as they are generated **Example:** ```python async for chunk in llm.generate_stream(messages, options): print(chunk, end="", flush=True) ``` ______________________________________________________________________ ## Data Models ### LLMMessage Represents a message in the conversation history. **Source**: [`src/xaibo/core/models/llm.py:40`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/llm.py#L40) #### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `role` | `LLMRole` | Yes | Message role (system, user, assistant, function) | | `content` | `List[LLMMessageContent]` | Yes | Message content (text, images, etc.) | | `name` | `str` | No | Optional name for the message sender | | `tool_calls` | `List[LLMFunctionCall]` | No | Function calls made by the assistant | | `tool_results` | `List[LLMFunctionResult]` | No | Results from function executions | #### Class Methods ______________________________________________________________________ ##### `system(content: str, name: Optional[str] = None) -> LLMMessage` Create a system message. ```python msg = LLMMessage.system("You are a helpful assistant") ``` ______________________________________________________________________ ##### `user(content: str, name: Optional[str] = None) -> LLMMessage` Create a user message. ```python msg = LLMMessage.user("Hello, how are you?") ``` ______________________________________________________________________ ##### `user_image(image_path: str, name: Optional[str] = None) -> LLMMessage` Create a user message with an image. ```python msg = LLMMessage.user_image("/path/to/image.jpg") ``` ______________________________________________________________________ ##### `assistant(content: str, name: Optional[str] = None) -> LLMMessage` Create an assistant message. ```python msg = LLMMessage.assistant("I'm doing well, thank you!") ``` ______________________________________________________________________ ##### `function(id: str, name: str, arguments: Dict[str, Any]) -> LLMMessage` Create a function call message. ```python msg = LLMMessage.function("call_123", "get_weather", {"city": "Paris"}) ``` ______________________________________________________________________ ##### `function_result(id: str, name: str, content: str) -> LLMMessage` Create a function result message. ```python msg = LLMMessage.function_result("call_123", "get_weather", "Sunny, 22°C") ``` ______________________________________________________________________ ### LLMRole Enumeration of message roles. **Source**: [`src/xaibo/core/models/llm.py:10`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/llm.py#L10) | Value | Description | | --- | --- | | `SYSTEM` | System instructions and context | | `USER` | User input messages | | `ASSISTANT` | Assistant responses | | `FUNCTION` | Function calls and results | ### LLMMessageContent Content within a message supporting multiple modalities. **Source**: [`src/xaibo/core/models/llm.py:34`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/llm.py#L34) #### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `type` | `LLMMessageContentType` | Yes | Content type (text or image) | | `text` | `str` | No | Text content (required if type is TEXT) | | `image` | `str` | No | Base64-encoded image data URI (required if type is IMAGE) | ### LLMMessageContentType Enumeration of content types. **Source**: [`src/xaibo/core/models/llm.py:17`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/llm.py#L17) | Value | Description | | --- | --- | | `TEXT` | Plain text content | | `IMAGE` | Image content (base64 data URI) | ### LLMOptions Configuration options for LLM generation. **Source**: [`src/xaibo/core/models/llm.py:90`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/llm.py#L90) #### Fields | Field | Type | Default | Description | | --- | --- | --- | --- | | `temperature` | `float` | `1.0` | Sampling temperature (0.0-2.0) | | `top_p` | `float` | `1.0` | Nucleus sampling parameter (0.0-1.0) | | `max_tokens` | `int` | `None` | Maximum tokens to generate | | `stop_sequences` | `List[str]` | `None` | Sequences that stop generation | | `functions` | `List[Tool]` | `None` | Available tools for function calling | | `vendor_specific` | `Dict[str, Any]` | `{}` | Provider-specific options | #### Validation - `temperature`: Must be between 0.0 and 2.0 - `top_p`: Must be between 0.0 and 1.0 ### LLMResponse Response from language model generation. **Source**: [`src/xaibo/core/models/llm.py:121`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/llm.py#L121) #### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `content` | `str` | Yes | Generated text content | | `tool_calls` | `List[LLMFunctionCall]` | No | Function calls requested by the model | | `usage` | `LLMUsage` | No | Token usage statistics | | `vendor_specific` | `Dict[str, Any]` | No | Provider-specific response data | #### Class Methods ______________________________________________________________________ ##### `merge(*responses: LLMResponse) -> LLMResponse` Merge multiple responses into a single response. ```python response1 = LLMResponse(content="Hello") response2 = LLMResponse(content="World") merged = LLMResponse.merge(response1, response2) print(merged.content) # "Hello\nWorld" ``` ______________________________________________________________________ ### LLMFunctionCall Function call information from the model. **Source**: [`src/xaibo/core/models/llm.py:22`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/llm.py#L22) #### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `id` | `str` | Yes | Unique identifier for the function call | | `name` | `str` | Yes | Name of the function to call | | `arguments` | `Dict[str, Any]` | Yes | Function arguments as key-value pairs | ### LLMFunctionResult Result from executing a function call. **Source**: [`src/xaibo/core/models/llm.py:28`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/llm.py#L28) #### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `id` | `str` | Yes | Identifier matching the original function call | | `name` | `str` | Yes | Name of the executed function | | `content` | `str` | Yes | String representation of the function result | ### LLMUsage Token usage statistics from the model. **Source**: [`src/xaibo/core/models/llm.py:114`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/llm.py#L114) #### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `prompt_tokens` | `int` | Yes | Tokens used in the input prompt | | `completion_tokens` | `int` | Yes | Tokens generated in the response | | `total_tokens` | `int` | Yes | Total tokens used (prompt + completion) | ## Implementation Example ```python from xaibo.core.protocols.llm import LLMProtocol from xaibo.core.models.llm import LLMMessage, LLMOptions, LLMResponse, LLMUsage class CustomLLM: """Example LLM implementation""" def __init__(self, config: dict): self.model = config.get("model", "default-model") self.api_key = config.get("api_key") self.base_url = config.get("base_url", "https://api.example.com") async def generate( self, messages: List[LLMMessage], options: Optional[LLMOptions] = None ) -> LLMResponse: # Convert messages to API format api_messages = self._convert_messages(messages) # Prepare request parameters params = { "model": self.model, "messages": api_messages } if options: if options.temperature is not None: params["temperature"] = options.temperature if options.max_tokens is not None: params["max_tokens"] = options.max_tokens if options.functions: params["tools"] = self._convert_tools(options.functions) # Make API request response = await self._api_request(params) # Convert response return LLMResponse( content=response["choices"][0]["message"]["content"], usage=LLMUsage( prompt_tokens=response["usage"]["prompt_tokens"], completion_tokens=response["usage"]["completion_tokens"], total_tokens=response["usage"]["total_tokens"] ) ) async def generate_stream( self, messages: List[LLMMessage], options: Optional[LLMOptions] = None ) -> AsyncIterator[str]: # Similar to generate but with streaming params = self._prepare_params(messages, options) params["stream"] = True async for chunk in self._api_stream(params): if chunk.get("choices") and chunk["choices"][0].get("delta"): content = chunk["choices"][0]["delta"].get("content", "") if content: yield content def _convert_messages(self, messages: List[LLMMessage]) -> List[dict]: """Convert LLMMessage objects to API format""" api_messages = [] for msg in messages: api_msg = {"role": msg.role.value} if msg.content: if len(msg.content) == 1 and msg.content[0].type == "text": api_msg["content"] = msg.content[0].text else: api_msg["content"] = [ self._convert_content(content) for content in msg.content ] if msg.tool_calls: api_msg["tool_calls"] = [ { "id": call.id, "type": "function", "function": { "name": call.name, "arguments": json.dumps(call.arguments) } } for call in msg.tool_calls ] api_messages.append(api_msg) return api_messages ``` ## Error Handling Implementations should handle and propagate appropriate exceptions: ```python class LLMError(Exception): """Base exception for LLM errors""" pass class LLMRateLimitError(LLMError): """Rate limit exceeded""" pass class LLMAuthenticationError(LLMError): """Authentication failed""" pass class LLMModelNotFoundError(LLMError): """Requested model not available""" pass ``` ## Testing Mock implementation for testing: ```python class MockLLM: def __init__(self, responses: List[str]): self.responses = responses self.call_count = 0 async def generate(self, messages, options=None): response = self.responses[self.call_count % len(self.responses)] self.call_count += 1 return LLMResponse( content=response, usage=LLMUsage(prompt_tokens=10, completion_tokens=5, total_tokens=15) ) async def generate_stream(self, messages, options=None): response = self.responses[self.call_count % len(self.responses)] self.call_count += 1 for word in response.split(): yield word + " " # Verify protocol compliance assert isinstance(MockLLM([]), LLMProtocol) ``` # Memory Protocol Specification The Memory Protocol defines interfaces for memory storage, retrieval, and vector-based semantic search in Xaibo agents. It provides a multi-layered architecture supporting chunking, embedding, indexing, and high-level memory operations. **Source**: [`src/xaibo/core/protocols/memory.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/memory.py) ## Protocol Hierarchy The memory system consists of four protocol layers: 1. **ChunkingProtocol**: Text segmentation for embedding 1. **EmbeddingProtocol**: Multi-modal embedding generation 1. **VectorIndexProtocol**: Vector storage and similarity search 1. **MemoryProtocol**: High-level memory operations ## ChunkingProtocol Protocol for splitting text into chunks suitable for embedding. ```python @runtime_checkable class ChunkingProtocol(Protocol): """Protocol for chunking text for embedding into a vector space""" async def chunk(self, text: str) -> List[str]: """Chunk text into smaller chunks for embedding""" ... ``` ### Methods ______________________________________________________________________ #### `chunk(text: str) -> List[str]` Split input text into smaller chunks optimized for embedding. **Parameters:** - `text` (`str`, required): Input text to be split into chunks **Returns:** - `List[str]`: List of text chunks suitable for embedding **Example:** ```python chunker = TokenChunker(window_size=512, window_overlap=50) chunks = await chunker.chunk("Long document text here...") print(f"Split into {len(chunks)} chunks") for i, chunk in enumerate(chunks): print(f"Chunk {i}: {chunk[:100]}...") ``` ______________________________________________________________________ ## EmbeddingProtocol Protocol for converting multiple modalities into vector embeddings. ```python @runtime_checkable class EmbeddingProtocol(Protocol): """Protocol for embedding multiple modalities into a vector space""" async def text_to_embedding(self, text: str) -> np.ndarray: """Convert text into vector embedding""" ... async def image_to_embedding(self, image_data: bytes) -> np.ndarray: """Convert image data into vector embedding""" ... async def audio_to_embedding(self, audio_data: bytes) -> np.ndarray: """Convert audio data into vector embedding""" ... ``` ### Methods ______________________________________________________________________ #### `text_to_embedding(text: str) -> np.ndarray` Convert text into a vector embedding. **Parameters:** - `text` (`str`, required): Input text to embed **Returns:** - `np.ndarray`: Vector embedding as NumPy array **Example:** ```python embedder = SentenceTransformerEmbedder(model_name="all-MiniLM-L6-v2") embedding = await embedder.text_to_embedding("Hello world") print(f"Embedding shape: {embedding.shape}") # (384,) ``` ______________________________________________________________________ #### `image_to_embedding(image_data: bytes) -> np.ndarray` Convert image data into a vector embedding. **Parameters:** - `image_data` (`bytes`, required): Raw image bytes to embed **Returns:** - `np.ndarray`: Vector embedding as NumPy array **Example:** ```python with open("image.jpg", "rb") as f: image_data = f.read() embedding = await embedder.image_to_embedding(image_data) print(f"Image embedding shape: {embedding.shape}") ``` ______________________________________________________________________ #### `audio_to_embedding(audio_data: bytes) -> np.ndarray` Convert audio data into a vector embedding. **Parameters:** - `audio_data` (`bytes`, required): Raw audio bytes to embed **Returns:** - `np.ndarray`: Vector embedding as NumPy array **Example:** ```python with open("audio.wav", "rb") as f: audio_data = f.read() embedding = await embedder.audio_to_embedding(audio_data) print(f"Audio embedding shape: {embedding.shape}") ``` ______________________________________________________________________ ## VectorIndexProtocol Protocol for indexing and searching vector embeddings. ```python @runtime_checkable class VectorIndexProtocol(Protocol): """Protocol for indexing and searching a vector space given a query vector""" async def add_vectors(self, vectors: List[np.ndarray], attributes: Optional[List[dict]] = None) -> None: """Add vectors to the index with optional attributes""" ... async def search(self, query_vector: np.ndarray, k: int = 10) -> List[VectorSearchResult]: """Search for similar vectors given a query vector""" ... ``` ### Methods ______________________________________________________________________ #### `add_vectors(vectors: List[np.ndarray], attributes: Optional[List[dict]] = None) -> None` Add vectors to the index with optional metadata. **Parameters:** - `vectors` (`List[np.ndarray]`, required): List of vector embeddings to add to index - `attributes` (`Optional[List[dict]]`, optional): Optional list of attribute dictionaries, one per vector **Example:** ```python vectors = [embedding1, embedding2, embedding3] attributes = [ {"source": "document1.txt", "page": 1}, {"source": "document1.txt", "page": 2}, {"source": "document2.txt", "page": 1} ] await vector_index.add_vectors(vectors, attributes) ``` ______________________________________________________________________ #### `search(query_vector: np.ndarray, k: int = 10) -> List[VectorSearchResult]` Search for similar vectors using a query vector. **Parameters:** - `query_vector` (`np.ndarray`, required): Vector embedding to search for - `k` (`int`, optional): Number of results to return (default: 10) **Returns:** - `List[VectorSearchResult]`: List of search results with similarity scores and attributes **Example:** ```python query_embedding = await embedder.text_to_embedding("search query") results = await vector_index.search(query_embedding, k=5) for result in results: print(f"ID: {result.vector_id}, Score: {result.similarity_score}") print(f"Attributes: {result.attributes}") ``` ______________________________________________________________________ ## MemoryProtocol High-level protocol for memory storage and retrieval operations. ```python @runtime_checkable class MemoryProtocol(Protocol): """Protocol for modules providing memory functionality""" async def store_memory(self, text: str, attributes: Optional[dict] = None) -> str: """Store a new memory""" ... async def get_memory(self, memory_id: str) -> Optional[dict]: """Retrieve a specific memory by ID""" ... async def search_memory(self, query: str, k: int = 10) -> List[MemorySearchResult]: """Search memories semantically""" ... async def list_memories(self) -> List[dict]: """List all stored memories""" ... async def delete_memory(self, memory_id: str) -> bool: """Delete a memory by ID""" ... async def update_memory(self, memory_id: str, text: str, attributes: Optional[dict] = None) -> bool: """Update an existing memory""" ... ``` ### Methods ______________________________________________________________________ #### `store_memory(text: str, attributes: Optional[dict] = None) -> str` Store a new memory with optional metadata. **Parameters:** - `text` (`str`, required): Text content to store - `attributes` (`Optional[dict]`, optional): Optional metadata attributes **Returns:** - `str`: ID of stored memory **Example:** ```python memory_id = await memory.store_memory( "Important meeting notes from today", attributes={ "date": "2024-01-15", "type": "meeting", "participants": ["Alice", "Bob"] } ) print(f"Stored memory with ID: {memory_id}") ``` ______________________________________________________________________ #### `get_memory(memory_id: str) -> Optional[dict]` Retrieve a specific memory by its ID. **Parameters:** - `memory_id` (`str`, required): ID of memory to retrieve **Returns:** - `Optional[dict]`: Memory data if found, None if not found **Example:** ```python memory_data = await memory.get_memory("mem_123") if memory_data: print(f"Content: {memory_data['content']}") print(f"Attributes: {memory_data['attributes']}") else: print("Memory not found") ``` ______________________________________________________________________ #### `search_memory(query: str, k: int = 10) -> List[MemorySearchResult]` Search memories using semantic similarity. **Parameters:** - `query` (`str`, required): Search query text - `k` (`int`, optional): Number of results to return (default: 10) **Returns:** - `List[MemorySearchResult]`: List of search results with memory content and similarity scores **Example:** ```python results = await memory.search_memory("meeting notes", k=5) for result in results: print(f"Memory ID: {result.memory_id}") print(f"Content: {result.content}") print(f"Similarity: {result.similarity_score}") print(f"Attributes: {result.attributes}") ``` ______________________________________________________________________ #### `list_memories() -> List[dict]` List all stored memories. **Returns:** - `List[dict]`: List of all memory entries **Example:** ```python all_memories = await memory.list_memories() print(f"Total memories: {len(all_memories)}") for mem in all_memories: print(f"ID: {mem['id']}, Content: {mem['content'][:50]}...") ``` ______________________________________________________________________ #### `delete_memory(memory_id: str) -> bool` Delete a memory by its ID. **Parameters:** - `memory_id` (`str`, required): ID of memory to delete **Returns:** - `bool`: True if deleted, False if not found **Example:** ```python deleted = await memory.delete_memory("mem_123") if deleted: print("Memory deleted successfully") else: print("Memory not found") ``` ______________________________________________________________________ #### `update_memory(memory_id: str, text: str, attributes: Optional[dict] = None) -> bool` Update an existing memory with new content and attributes. **Parameters:** - `memory_id` (`str`, required): ID of memory to update - `text` (`str`, required): New text content - `attributes` (`Optional[dict]`, optional): Optional new metadata attributes **Returns:** - `bool`: True if updated, False if not found **Example:** ```python updated = await memory.update_memory( "mem_123", "Updated meeting notes with action items", attributes={"status": "updated", "action_items": 3} ) if updated: print("Memory updated successfully") else: print("Memory not found") ``` ______________________________________________________________________ ## Data Models ### MemorySearchResult Result from memory search operations. **Source**: [`src/xaibo/core/protocols/memory.py:6`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/memory.py#L6) #### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `memory_id` | `str` | Yes | Unique identifier for the memory | | `content` | `str` | Yes | Text content of the memory | | `similarity_score` | `float` | Yes | Similarity score (0.0 to 1.0) | | `attributes` | `Dict[str, Any]` | No | Optional metadata attributes | ### VectorSearchResult Result from vector index search operations. **Source**: [`src/xaibo/core/protocols/memory.py:14`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/memory.py#L14) #### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `vector_id` | `str` | Yes | Unique identifier for the vector | | `similarity_score` | `float` | Yes | Similarity score (0.0 to 1.0) | | `attributes` | `Dict[str, Any]` | No | Optional metadata attributes | ## Implementation Example ```python from xaibo.core.protocols.memory import MemoryProtocol, ChunkingProtocol, EmbeddingProtocol, VectorIndexProtocol from xaibo.core.protocols.memory import MemorySearchResult, VectorSearchResult import uuid import numpy as np from typing import Dict, List, Optional, Any class SimpleMemorySystem: """Example implementation combining all memory protocols""" def __init__( self, chunker: ChunkingProtocol, embedder: EmbeddingProtocol, vector_index: VectorIndexProtocol ): self.chunker = chunker self.embedder = embedder self.vector_index = vector_index self.memories: Dict[str, dict] = {} self.chunk_to_memory: Dict[str, str] = {} async def store_memory(self, text: str, attributes: Optional[dict] = None) -> str: """Store a new memory with chunking and embedding""" memory_id = str(uuid.uuid4()) # Store memory metadata self.memories[memory_id] = { "id": memory_id, "content": text, "attributes": attributes or {}, "chunks": [] } # Chunk the text chunks = await self.chunker.chunk(text) # Embed each chunk vectors = [] chunk_attributes = [] for i, chunk in enumerate(chunks): chunk_id = f"{memory_id}_chunk_{i}" self.chunk_to_memory[chunk_id] = memory_id self.memories[memory_id]["chunks"].append(chunk_id) # Create embedding embedding = await self.embedder.text_to_embedding(chunk) vectors.append(embedding) # Prepare attributes chunk_attrs = { "memory_id": memory_id, "chunk_id": chunk_id, "chunk_index": i, "chunk_text": chunk, **(attributes or {}) } chunk_attributes.append(chunk_attrs) # Add to vector index await self.vector_index.add_vectors(vectors, chunk_attributes) return memory_id async def get_memory(self, memory_id: str) -> Optional[dict]: """Retrieve a specific memory by ID""" return self.memories.get(memory_id) async def search_memory(self, query: str, k: int = 10) -> List[MemorySearchResult]: """Search memories using semantic similarity""" # Embed the query query_embedding = await self.embedder.text_to_embedding(query) # Search vector index vector_results = await self.vector_index.search(query_embedding, k * 2) # Get more to deduplicate # Group results by memory and take best score per memory memory_scores: Dict[str, float] = {} memory_chunks: Dict[str, List[str]] = {} for result in vector_results: memory_id = result.attributes["memory_id"] chunk_text = result.attributes["chunk_text"] if memory_id not in memory_scores or result.similarity_score > memory_scores[memory_id]: memory_scores[memory_id] = result.similarity_score if memory_id not in memory_chunks: memory_chunks[memory_id] = [] memory_chunks[memory_id].append(chunk_text) # Create memory search results results = [] for memory_id in sorted(memory_scores.keys(), key=lambda x: memory_scores[x], reverse=True)[:k]: memory_data = self.memories[memory_id] results.append(MemorySearchResult( memory_id=memory_id, content=memory_data["content"], similarity_score=memory_scores[memory_id], attributes=memory_data["attributes"] )) return results async def list_memories(self) -> List[dict]: """List all stored memories""" return list(self.memories.values()) async def delete_memory(self, memory_id: str) -> bool: """Delete a memory and its chunks""" if memory_id not in self.memories: return False # Clean up chunk mappings memory_data = self.memories[memory_id] for chunk_id in memory_data.get("chunks", []): self.chunk_to_memory.pop(chunk_id, None) # Remove memory del self.memories[memory_id] # Note: In a real implementation, you'd also remove vectors from the index return True async def update_memory(self, memory_id: str, text: str, attributes: Optional[dict] = None) -> bool: """Update an existing memory""" if memory_id not in self.memories: return False # Delete old memory await self.delete_memory(memory_id) # Store updated memory with same ID self.memories[memory_id] = { "id": memory_id, "content": text, "attributes": attributes or {}, "chunks": [] } # Re-chunk and re-embed chunks = await self.chunker.chunk(text) vectors = [] chunk_attributes = [] for i, chunk in enumerate(chunks): chunk_id = f"{memory_id}_chunk_{i}" self.chunk_to_memory[chunk_id] = memory_id self.memories[memory_id]["chunks"].append(chunk_id) embedding = await self.embedder.text_to_embedding(chunk) vectors.append(embedding) chunk_attrs = { "memory_id": memory_id, "chunk_id": chunk_id, "chunk_index": i, "chunk_text": chunk, **(attributes or {}) } chunk_attributes.append(chunk_attrs) await self.vector_index.add_vectors(vectors, chunk_attributes) return True # Verify protocol compliance assert isinstance(SimpleMemorySystem(None, None, None), MemoryProtocol) ``` ## Testing Mock implementations for testing: ```python class MockChunker: def __init__(self, chunk_size: int = 100): self.chunk_size = chunk_size async def chunk(self, text: str) -> List[str]: # Simple character-based chunking chunks = [] for i in range(0, len(text), self.chunk_size): chunks.append(text[i:i + self.chunk_size]) return chunks class MockEmbedder: def __init__(self, dimension: int = 384): self.dimension = dimension async def text_to_embedding(self, text: str) -> np.ndarray: # Generate deterministic embedding based on text hash import hashlib hash_value = int(hashlib.md5(text.encode()).hexdigest(), 16) np.random.seed(hash_value % (2**32)) return np.random.normal(0, 1, self.dimension).astype(np.float32) async def image_to_embedding(self, image_data: bytes) -> np.ndarray: return await self.text_to_embedding(str(len(image_data))) async def audio_to_embedding(self, audio_data: bytes) -> np.ndarray: return await self.text_to_embedding(str(len(audio_data))) class MockVectorIndex: def __init__(self): self.vectors: List[np.ndarray] = [] self.attributes: List[dict] = [] async def add_vectors(self, vectors: List[np.ndarray], attributes: Optional[List[dict]] = None) -> None: self.vectors.extend(vectors) if attributes: self.attributes.extend(attributes) else: self.attributes.extend([{}] * len(vectors)) async def search(self, query_vector: np.ndarray, k: int = 10) -> List[VectorSearchResult]: if not self.vectors: return [] # Calculate cosine similarity similarities = [] for i, vector in enumerate(self.vectors): similarity = np.dot(query_vector, vector) / (np.linalg.norm(query_vector) * np.linalg.norm(vector)) similarities.append((i, similarity)) # Sort by similarity and return top k similarities.sort(key=lambda x: x[1], reverse=True) results = [] for i, (vector_idx, score) in enumerate(similarities[:k]): results.append(VectorSearchResult( vector_id=str(vector_idx), similarity_score=float(score), attributes=self.attributes[vector_idx] )) return results # Verify protocol compliance assert isinstance(MockChunker(), ChunkingProtocol) assert isinstance(MockEmbedder(), EmbeddingProtocol) assert isinstance(MockVectorIndex(), VectorIndexProtocol) ``` ## Best Practices ### Memory Design 1. **Chunking Strategy**: Choose appropriate chunk sizes for your embedding model 1. **Metadata**: Store rich metadata for filtering and context 1. **Deduplication**: Handle duplicate or similar content appropriately 1. **Versioning**: Consider versioning for updated memories 1. **Cleanup**: Implement proper cleanup for deleted memories ### Performance Optimization 1. **Batch Operations**: Process multiple items together when possible 1. **Caching**: Cache frequently accessed embeddings 1. **Indexing**: Use efficient vector indexing algorithms 1. **Lazy Loading**: Load large memories on demand 1. **Compression**: Consider embedding compression for storage ### Error Handling 1. **Graceful Degradation**: Handle embedding failures gracefully 1. **Validation**: Validate input text and parameters 1. **Resource Limits**: Implement memory and storage limits 1. **Retry Logic**: Add retry logic for transient failures 1. **Monitoring**: Monitor memory usage and performance # Response Protocol Specification The Response Protocol defines the interface for sending responses from Xaibo agents. It provides standardized methods for delivering text, multimedia content, and complex responses with attachments. **Source**: [`src/xaibo/core/protocols/response.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/response.py) ## ResponseProtocol The core protocol interface for response handling implementations. ```python @runtime_checkable class ResponseProtocol(Protocol): """Protocol for sending responses""" async def get_response(self) -> Response: """Get the current response object""" ... async def respond_text(self, response: str) -> None: """Send a response""" ... async def respond_image(self, iolike: BinaryIO) -> None: """Send an image response""" ... async def respond_audio(self, iolike: BinaryIO) -> None: """Send an audio response""" ... async def respond_file(self, iolike: BinaryIO) -> None: """Send a file response""" ... async def respond(self, response: Response) -> None: """Send a complex response containing text and/or file attachments""" ... ``` ### Methods ______________________________________________________________________ #### `get_response() -> Response` Retrieve the current response object containing accumulated text and attachments. **Returns:** - [`Response`](#response): The current response object with text content and file attachments **Example:** ```python current_response = await response_handler.get_response() if current_response.text: print(f"Response text: {current_response.text}") if current_response.attachments: print(f"Attachments: {len(current_response.attachments)}") ``` ______________________________________________________________________ #### `respond_text(response) -> None` Send a text response to the user. **Parameters:** - `response` (`str`, required): The response text to send **Example:** ```python await response_handler.respond_text("Hello! How can I help you today?") ``` ______________________________________________________________________ #### `respond_image(iolike) -> None` Send an image response to the user. **Parameters:** - `iolike` (`BinaryIO`, required): IO object containing the image data **Example:** ```python with open("chart.png", "rb") as image_file: await response_handler.respond_image(image_file) ``` ______________________________________________________________________ #### `respond_audio(iolike) -> None` Send an audio response to the user. **Parameters:** - `iolike` (`BinaryIO`, required): IO object containing the audio data **Example:** ```python with open("speech.mp3", "rb") as audio_file: await response_handler.respond_audio(audio_file) ``` ______________________________________________________________________ #### `respond_file(iolike) -> None` Send a file response to the user. **Parameters:** - `iolike` (`BinaryIO`, required): IO object containing the file data **Example:** ```python with open("report.pdf", "rb") as file: await response_handler.respond_file(file) ``` ______________________________________________________________________ #### `respond(response) -> None` Send a complex response containing text and/or file attachments. **Parameters:** - `response` ([`Response`](#response), required): Response object containing text and attachments **Example:** ```python attachments = [ FileAttachment(image_data, FileType.IMAGE), FileAttachment(audio_data, FileType.AUDIO) ] complex_response = Response( text="Here's your analysis with supporting materials:", attachments=attachments ) await response_handler.respond(complex_response) ``` ______________________________________________________________________ ## Data Structures ### Response Model for responses that can include text and file attachments. **Source**: [`src/xaibo/core/models/response.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/response.py) ```python class Response: """Model for responses that can include text and file attachments""" text: Optional[str] = None attachments: List[FileAttachment] = [] def __init__( self, text: Optional[str] = None, attachments: Optional[List[FileAttachment]] = None ) -> None: ... ``` **Attributes:** - `text` (`Optional[str]`): Text content of the response - `attachments` (`List[FileAttachment]`): List of file attachments ### FileAttachment Model for file attachments in responses. ```python class FileAttachment: """Model for file attachments in responses""" content: BinaryIO type: FileType def __init__(self, content: BinaryIO, type: FileType) -> None: ... ``` **Attributes:** - `content` (`BinaryIO`): IO object containing the file data - `type` ([`FileType`](#filetype)): Type classification of the file ### FileType Enumeration for different types of file attachments. ```python class FileType(Enum): """Enum for different types of file attachments""" IMAGE = "image" AUDIO = "audio" FILE = "file" ``` **Values:** - `IMAGE`: Image file attachments (PNG, JPEG, GIF, etc.) - `AUDIO`: Audio file attachments (MP3, WAV, etc.) - `FILE`: Generic file attachments (PDF, documents, etc.) ## Implementation Requirements ### Asynchronous Operations All protocol methods must be implemented as asynchronous functions using `async`/`await` syntax. ### Error Handling Implementations must handle the following error conditions: - **Invalid file data**: When `BinaryIO` objects contain invalid or corrupted data - **Unsupported file types**: When file types are not supported by the response handler - **Network failures**: When response delivery fails due to connectivity issues - **Resource limits**: When file sizes exceed implementation-specific limits ### Thread Safety Response protocol implementations must be thread-safe when accessed from multiple concurrent tasks. ## Usage Patterns in Orchestrator Modules ### Basic Text Response ```python class SimpleOrchestrator: def __init__(self, response: ResponseProtocol): self.response = response async def handle_message(self, message: str): result = await self.process_message(message) await self.response.respond_text(result) ``` ### Multi-Modal Response ```python class AnalysisOrchestrator: def __init__(self, response: ResponseProtocol): self.response = response async def generate_report(self, data): # Generate text analysis analysis_text = await self.analyze_data(data) # Generate visualization chart_data = await self.create_chart(data) # Send combined response attachments = [FileAttachment(chart_data, FileType.IMAGE)] response = Response(text=analysis_text, attachments=attachments) await self.response.respond(response) ``` ### Streaming Response Pattern ```python class StreamingOrchestrator: def __init__(self, response: ResponseProtocol): self.response = response async def process_long_task(self, request): # Send initial response await self.response.respond_text("Processing your request...") # Process and send intermediate results for step_result in self.process_steps(request): await self.response.respond_text(f"Step completed: {step_result}") # Send final response with attachments final_result = await self.finalize_processing() await self.response.respond(final_result) ``` ## Error Handling Specifications ### Exception Types Implementations should raise appropriate exceptions for error conditions: - `ValueError`: For invalid response content or malformed data - `IOError`: For file I/O related errors - `ConnectionError`: For network-related response delivery failures - `RuntimeError`: For general response handling failures ### Error Recovery Response handlers should implement graceful error recovery: 1. **Retry Logic**: Automatic retry for transient network failures 1. **Fallback Responses**: Alternative response methods when primary method fails 1. **Error Logging**: Comprehensive logging of error conditions for debugging 1. **Partial Delivery**: Ability to deliver partial responses when complete delivery fails ### Timeout Handling Response operations must implement appropriate timeouts: - **Text responses**: 30 seconds maximum - **File uploads**: 5 minutes maximum for large files - **Network operations**: Configurable timeout with reasonable defaults # Tools Protocol Specification The Tools Protocol defines the interface for tool providers in Xaibo agents. It standardizes tool discovery, execution, and result handling across different tool implementations. **Source**: [`src/xaibo/core/protocols/tools.py`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/tools.py) ## ToolProviderProtocol The core protocol interface for tool provider implementations. ```python @runtime_checkable class ToolProviderProtocol(Protocol): """Protocol for providing and executing tools""" async def list_tools(self) -> List[Tool]: """List all available tools provided by this provider""" ... async def execute_tool(self, tool_name: str, parameters: Dict[str, Any]) -> ToolResult: """Execute a tool with the given parameters""" ... ``` ### Methods ______________________________________________________________________ #### `list_tools() -> List[Tool]` Retrieve all tools available from this provider. **Returns:** - `List[Tool]`: List of tool definitions with names, descriptions, and parameter schemas **Example:** ```python tools = await tool_provider.list_tools() for tool in tools: print(f"{tool.name}: {tool.description}") for param_name, param in tool.parameters.items(): print(f" {param_name} ({param.type}): {param.description}") ``` ______________________________________________________________________ #### `execute_tool(tool_name: str, parameters: Dict[str, Any]) -> ToolResult` Execute a specific tool with provided parameters. **Parameters:** - `tool_name` (`str`, required): Name of the tool to execute - `parameters` (`Dict[str, Any]`, required): Tool parameters as key-value pairs **Returns:** - `ToolResult`: Execution result with success status, result data, or error information **Raises:** - `ToolNotFoundError`: If the specified tool is not available - `ToolExecutionError`: If tool execution fails **Example:** ```python result = await tool_provider.execute_tool( "get_weather", {"city": "Paris", "units": "metric"} ) if result.success: print(f"Weather data: {result.result}") else: print(f"Error: {result.error}") ``` ______________________________________________________________________ ## Data Models ### Tool Definition of an executable tool with its interface specification. **Source**: [`src/xaibo/core/models/tools.py:15`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/tools.py#L15) #### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `name` | `str` | Yes | Unique identifier for the tool | | `description` | `str` | Yes | Human-readable description of tool functionality | | `parameters` | `Dict[str, ToolParameter]` | No | Parameter definitions (default: empty dict) | #### Example ```python tool = Tool( name="get_weather", description="Get current weather information for a city", parameters={ "city": ToolParameter( type="string", description="Name of the city", required=True ), "units": ToolParameter( type="string", description="Temperature units", required=False, default="celsius", enum=["celsius", "fahrenheit", "kelvin"] ) } ) ``` ### ToolParameter Parameter definition for tool inputs. **Source**: [`src/xaibo/core/models/tools.py:6`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/tools.py#L6) #### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `type` | `str` | Yes | Parameter type (string, number, boolean, object, array) | | `description` | `str` | No | Human-readable parameter description | | `required` | `bool` | No | Whether parameter is required (default: False) | | `default` | `Any` | No | Default value if parameter not provided | | `enum` | `List[str]` | No | List of allowed values for string parameters | #### Supported Types | Type | Description | Example Values | | --- | --- | --- | | `string` | Text values | `"hello"`, `"world"` | | `number` | Numeric values | `42`, `3.14`, `-10` | | `integer` | Integer values | `42`, `-10`, `0` | | `boolean` | Boolean values | `true`, `false` | | `object` | JSON objects | `{"key": "value"}` | | `array` | JSON arrays | `[1, 2, 3]`, `["a", "b"]` | #### Example ```python param = ToolParameter( type="string", description="The city name to get weather for", required=True ) enum_param = ToolParameter( type="string", description="Temperature units", required=False, default="celsius", enum=["celsius", "fahrenheit", "kelvin"] ) number_param = ToolParameter( type="number", description="Maximum temperature threshold", required=False, default=30.0 ) ``` ### ToolResult Result of tool execution with success status and data. **Source**: [`src/xaibo/core/models/tools.py:22`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/tools.py#L22) #### Fields | Field | Type | Required | Description | | --- | --- | --- | --- | | `success` | `bool` | Yes | Whether the tool execution succeeded | | `result` | `Any` | No | Tool execution result data (if successful) | | `error` | `str` | No | Error message (if execution failed) | #### Success Result ```python result = ToolResult( success=True, result={ "temperature": 22.5, "humidity": 65, "conditions": "partly cloudy" } ) ``` #### Error Result ```python result = ToolResult( success=False, error="City 'InvalidCity' not found" ) ``` ## Implementation Example ```python from xaibo.core.protocols.tools import ToolProviderProtocol from xaibo.core.models.tools import Tool, ToolParameter, ToolResult import asyncio import json class WeatherToolProvider: """Example tool provider for weather information""" def __init__(self, config: dict): self.api_key = config.get("api_key") self.base_url = config.get("base_url", "https://api.weather.com") async def list_tools(self) -> List[Tool]: """Return available weather tools""" return [ Tool( name="get_current_weather", description="Get current weather conditions for a city", parameters={ "city": ToolParameter( type="string", description="City name", required=True ), "units": ToolParameter( type="string", description="Temperature units", required=False, default="celsius", enum=["celsius", "fahrenheit", "kelvin"] ) } ), Tool( name="get_weather_forecast", description="Get weather forecast for multiple days", parameters={ "city": ToolParameter( type="string", description="City name", required=True ), "days": ToolParameter( type="integer", description="Number of days to forecast", required=False, default=5 ) } ) ] async def execute_tool(self, tool_name: str, parameters: Dict[str, Any]) -> ToolResult: """Execute a weather tool""" try: if tool_name == "get_current_weather": return await self._get_current_weather(parameters) elif tool_name == "get_weather_forecast": return await self._get_weather_forecast(parameters) else: return ToolResult( success=False, error=f"Unknown tool: {tool_name}" ) except Exception as e: return ToolResult( success=False, error=f"Tool execution failed: {str(e)}" ) async def _get_current_weather(self, parameters: Dict[str, Any]) -> ToolResult: """Get current weather for a city""" city = parameters.get("city") units = parameters.get("units", "celsius") if not city: return ToolResult( success=False, error="City parameter is required" ) # Simulate API call await asyncio.sleep(0.1) # Simulate network delay # Mock weather data weather_data = { "city": city, "temperature": 22.5 if units == "celsius" else 72.5, "units": units, "humidity": 65, "conditions": "partly cloudy", "wind_speed": 10 } return ToolResult( success=True, result=weather_data ) async def _get_weather_forecast(self, parameters: Dict[str, Any]) -> ToolResult: """Get weather forecast for a city""" city = parameters.get("city") days = parameters.get("days", 5) if not city: return ToolResult( success=False, error="City parameter is required" ) # Simulate API call await asyncio.sleep(0.2) # Mock forecast data forecast = { "city": city, "forecast": [ { "date": f"2024-01-{i+1:02d}", "high": 25 + i, "low": 15 + i, "conditions": "sunny" if i % 2 == 0 else "cloudy" } for i in range(days) ] } return ToolResult( success=True, result=forecast ) # Verify protocol compliance assert isinstance(WeatherToolProvider({}), ToolProviderProtocol) ``` ## Error Handling Tool providers should handle errors gracefully and return appropriate ToolResult objects: ```python class ToolError(Exception): """Base exception for tool errors""" pass class ToolNotFoundError(ToolError): """Tool not found""" pass class ToolExecutionError(ToolError): """Tool execution failed""" pass class ToolParameterError(ToolError): """Invalid tool parameters""" pass # Example error handling in execute_tool async def execute_tool(self, tool_name: str, parameters: Dict[str, Any]) -> ToolResult: try: # Validate tool exists tools = await self.list_tools() tool_names = [t.name for t in tools] if tool_name not in tool_names: return ToolResult( success=False, error=f"Tool '{tool_name}' not found. Available tools: {tool_names}" ) # Validate parameters tool = next(t for t in tools if t.name == tool_name) validation_error = self._validate_parameters(tool, parameters) if validation_error: return ToolResult( success=False, error=validation_error ) # Execute tool result = await self._execute_tool_impl(tool_name, parameters) return ToolResult(success=True, result=result) except Exception as e: return ToolResult( success=False, error=f"Unexpected error: {str(e)}" ) ``` ## Testing Mock implementation for testing: ```python class MockToolProvider: """Mock tool provider for testing""" def __init__(self, tools: List[Tool], results: Dict[str, Any]): self.tools = tools self.results = results self.call_history = [] async def list_tools(self) -> List[Tool]: return self.tools async def execute_tool(self, tool_name: str, parameters: Dict[str, Any]) -> ToolResult: self.call_history.append((tool_name, parameters)) if tool_name in self.results: result_data = self.results[tool_name] if isinstance(result_data, Exception): return ToolResult(success=False, error=str(result_data)) else: return ToolResult(success=True, result=result_data) else: return ToolResult(success=False, error=f"Tool {tool_name} not mocked") # Usage in tests mock_tools = [ Tool(name="test_tool", description="A test tool", parameters={}) ] mock_results = { "test_tool": {"message": "success"} } mock_provider = MockToolProvider(mock_tools, mock_results) assert isinstance(mock_provider, ToolProviderProtocol) # Test execution result = await mock_provider.execute_tool("test_tool", {}) assert result.success assert result.result == {"message": "success"} ``` ## Integration with LLM Function Calling Tools integrate with LLM function calling through the LLMOptions: ```python # Get tools from provider tools = await tool_provider.list_tools() # Use in LLM options options = LLMOptions( temperature=0.7, functions=tools # Tools available for function calling ) # Generate response with tools response = await llm.generate(messages, options) # Execute any tool calls if response.tool_calls: for call in response.tool_calls: result = await tool_provider.execute_tool(call.name, call.arguments) # Add result to conversation history messages.append(LLMMessage.function_result(call.id, call.name, str(result.result))) ``` ## Best Practices ### Tool Design 1. **Clear Names**: Use descriptive, action-oriented tool names 1. **Comprehensive Descriptions**: Explain what the tool does and when to use it 1. **Parameter Validation**: Validate all input parameters 1. **Error Handling**: Return meaningful error messages 1. **Idempotency**: Tools should be safe to call multiple times ### Parameter Definitions 1. **Type Safety**: Use appropriate parameter types 1. **Required Fields**: Mark required parameters explicitly 1. **Default Values**: Provide sensible defaults where possible 1. **Enums**: Use enums for limited choice parameters 1. **Documentation**: Include clear parameter descriptions ### Result Handling 1. **Consistent Format**: Use consistent result structures 1. **Success Indication**: Always indicate success/failure clearly 1. **Error Details**: Provide actionable error messages 1. **Serializable Results**: Ensure results can be JSON serialized 1. **Size Limits**: Keep results reasonably sized for LLM context # Explanation # Understanding Xaibo Welcome to the explanation section of Xaibo's documentation. This section is designed to deepen your understanding of the framework's design philosophy, architectural decisions, and core concepts. Rather than teaching you how to use specific features, these discussions illuminate *why* Xaibo works the way it does and help you develop a mental model of the framework. ## What Makes Xaibo Different Xaibo represents a departure from traditional AI agent frameworks. Where many frameworks focus on quick prototyping or specific use cases, Xaibo prioritizes long-term maintainability, testability, and extensibility. This philosophical difference permeates every aspect of the framework's design. The framework emerged from the recognition that AI systems, like any complex software, benefit from well-established software engineering principles. However, AI systems also present unique challenges: they're inherently non-deterministic, they integrate with rapidly evolving external services, and they often require complex orchestration between multiple components. ## The Architecture Philosophy Xaibo's architecture is built around three fundamental principles that work together to create a robust foundation for AI systems: **Protocol-Driven Design**: Instead of concrete dependencies, components interact through well-defined interfaces. This creates natural boundaries that make the system more modular and testable. **Transparent Observability**: Every interaction between components is automatically captured and made visible. This isn't just logging, it's a comprehensive view into how your agent thinks and operates. **Dependency Injection**: Components explicitly declare what they need, allowing the framework to wire them together automatically while maintaining flexibility for testing and configuration. ## Understanding Through Exploration The explanations in this section are organized around different aspects of understanding Xaibo: ### Architecture Deep Dives - **[Protocol-Driven Architecture](architecture/protocols/)**: How protocols create clean boundaries and enable flexibility - **[Dependency Injection](architecture/dependency-injection/)**: Why explicit dependencies lead to better software - **[Transparent Proxies](architecture/transparent-proxies/)**: How observability is built into the framework's core ### Core Concepts - **[Modules vs Protocols](concepts/modules-vs-protocols/)**: Understanding the relationship between implementation and interface - **[Exchange System](concepts/exchange-system/)**: How components are wired together automatically - **[Event System](concepts/event-system/)**: The comprehensive observability layer ### Design Decisions - **[Modularity](design/modularity/)**: Why Xaibo chose composition over inheritance - **[Testability](design/testability/)**: How the architecture enables superior testing strategies - **[Extensibility](design/extensibility/)**: Building systems that grow with your needs ## Reading These Explanations These documents are designed to be read at leisure, away from your code editor. They explore concepts from multiple angles, consider alternatives, and sometimes discuss approaches that Xaibo explicitly chose not to take. This broader perspective helps you understand not just what Xaibo does, but why it does it that way. Think of these explanations as conversations about software architecture, using Xaibo as a concrete example of how abstract principles can be applied to solve real problems in AI system development. ## The Bigger Picture Xaibo's design reflects broader trends in software engineering: the move toward microservices, the emphasis on observability, and the recognition that complex systems require careful architectural planning. By understanding these principles in the context of AI agents, you'll be better equipped to build systems that are not just functional, but maintainable and extensible over time. The framework doesn't just solve today's problems, it provides a foundation for solving tomorrow's problems too. # Understanding Dependency Injection in Xaibo Dependency injection is one of those concepts that sounds more complicated than it actually is. At its core, it's about making dependencies explicit rather than hidden, and letting a framework handle the tedious work of wiring components together. In Xaibo, dependency injection isn't just a convenience, it's fundamental to how the framework enables modularity, testability, and flexibility. ## The Problem of Hidden Dependencies Consider a traditional approach to building an AI agent. You might write an orchestrator class that looks something like this: ```python class TraditionalOrchestrator: def __init__(self): self.llm = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) self.memory = VectorMemory(embedder=OpenAIEmbedder()) self.tools = PythonTools(package="my_tools") ``` This seems straightforward, but it hides several problems. The orchestrator is making decisions about which specific implementations to use, how to configure them, and where to get configuration values. It's tightly coupled to OpenAI, to a specific memory implementation, and to a particular way of loading tools. More subtly, the dependencies aren't visible from the outside. If you want to test this orchestrator, you need to understand its internal implementation to know what external services it will try to contact. If you want to use a different LLM provider, you need to modify the orchestrator's code. ## Making Dependencies Explicit Xaibo takes a different approach. Components declare their dependencies explicitly in their constructor signatures: ```python class StressingToolUser: def __init__(self, response: ResponseProtocol, llm: LLMProtocol, tool_provider: ToolProviderProtocol, history: ConversationHistoryProtocol, config: dict = None): self.response = response self.llm = llm self.tool_provider = tool_provider self.history = history # ... initialization logic ``` This simple change has profound implications. The orchestrator no longer makes assumptions about which specific implementations it will receive. It declares that it needs something that can fulfill the `LLMProtocol`, but it doesn't care whether that's OpenAI, Anthropic, or a mock implementation for testing. The dependencies are now visible in the type signature. Anyone reading the code can immediately see what this component needs to function. Testing becomes straightforward because you can inject mock implementations that fulfill the same protocols. ## The Wiring Problem Of course, making dependencies explicit creates a new problem: something needs to actually provide those dependencies. This is where Xaibo's exchange system comes in. The [`Exchange`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/exchange.py) class acts as a sophisticated dependency injection container that automatically wires components together based on their declared needs and the available implementations. When you configure an agent, you specify which modules to instantiate: ```yaml modules: - module: xaibo.primitives.modules.llm.OpenAILLM id: llm config: model: gpt-4 - module: xaibo.primitives.modules.orchestrator.StressingToolUser id: orchestrator config: max_thoughts: 10 ``` The exchange system analyzes the constructor signatures of these modules, identifies their dependencies, and automatically creates the necessary connections. The `StressingToolUser` needs an `LLMProtocol`, and the `OpenAILLM` provides one, so the exchange system injects the LLM instance into the orchestrator. ## Automatic Dependency Resolution This automatic wiring is more sophisticated than it might initially appear. The exchange system handles several complex scenarios: **Type-based matching**: Dependencies are matched based on the protocol types they implement. If a module needs an `LLMProtocol` and only one module provides it, the connection is made automatically. **List dependencies**: Some modules need multiple implementations of the same protocol. The exchange system can inject lists of implementations, enabling modules like [`ToolCollector`](../../../reference/modules/tools/#toolcollector) to aggregate multiple tool providers into a unified interface. **Dependency ordering**: The exchange system automatically determines the order in which modules should be instantiated, ensuring that dependencies are available before the modules that need them. **Circular dependency detection**: The system can detect and report circular dependencies that would prevent successful instantiation. ## Configuration Flexibility The automatic wiring works well for simple cases, but complex systems often need explicit control over how dependencies are resolved. Xaibo's exchange configuration provides this control: ```yaml exchange: - module: orchestrator protocol: LLMProtocol provider: llm - module: orchestrator protocol: ToolsProtocol provider: [python-tools, mcp-tools] ``` This explicit configuration overrides the automatic matching, allowing you to specify exactly which implementations should be used for which dependencies. This is particularly useful when you have multiple implementations of the same protocol and need to control which one goes where. ## Testing Through Injection The real power of dependency injection becomes apparent when testing. Instead of trying to mock external services or intercept network calls, you can simply inject test implementations: ```python # Create test implementations mock_llm = MockLLM(responses=["Test response"]) mock_tools = MockTools(available_tools=[]) # Create exchange with test dependencies exchange = Exchange( config=agent_config, override_config=ConfigOverrides( instances={ 'llm': mock_llm, 'tools': mock_tools } ) ) ``` The orchestrator receives these mock implementations through the same dependency injection mechanism used in production. It doesn't know or care that it's running in a test environment, it just uses whatever implementations it receives. This approach makes tests faster, more reliable, and more focused. You can test the orchestrator's logic without depending on external services, network connectivity, or API rate limits. ## The Inversion of Control Dependency injection represents a broader principle called "inversion of control." Instead of components controlling their own dependencies, the control is inverted, an external system (the exchange) controls what dependencies each component receives. This inversion has several benefits: **Flexibility**: The same component can work with different implementations depending on how it's configured. **Testability**: Test implementations can be injected without modifying the component being tested. **Separation of concerns**: Components focus on their core logic rather than on managing their dependencies. **Configuration centralization**: Dependency relationships are managed in one place rather than scattered throughout the codebase. ## Lifecycle Management Xaibo's dependency injection system also handles component lifecycle management. When the exchange instantiates modules, it ensures they're created in the correct order and that their dependencies are properly initialized. This eliminates a whole class of initialization bugs that can occur when components try to use dependencies that haven't been properly set up. The exchange also provides access to components through the [`Proxy`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/exchange.py) system, which adds observability and event handling without requiring changes to the underlying components. ## Comparison with Other Approaches Different frameworks handle dependency management in different ways. Some use service locator patterns, where components ask a central registry for their dependencies. Others use factory patterns, where specialized classes are responsible for creating and configuring components. Xaibo's approach combines the best aspects of these patterns while avoiding their pitfalls. Like service locators, it provides centralized dependency management. Like factories, it handles the complexity of component creation. But unlike either pattern, it makes dependencies explicit in component signatures, which improves code clarity and testability. ## The Cost of Explicitness Making dependencies explicit does have costs. Component constructors become more verbose, and you need to understand the exchange system to configure complex dependency relationships. There's also a conceptual overhead, developers need to think about interfaces and implementations rather than just concrete objects. However, these costs are front-loaded. Once you understand the dependency injection pattern, it makes the system more predictable and easier to work with. The explicitness that seems like overhead when writing code becomes invaluable when debugging, testing, or extending the system. ## Dependency Injection in AI Systems Dependency injection is particularly valuable in AI systems because of their inherent complexity and rapid evolution. AI systems typically integrate multiple external services, each with their own configuration requirements, error modes, and performance characteristics. By making these dependencies explicit and injectable, Xaibo enables you to: - **Switch providers easily**: Move from OpenAI to Anthropic without rewriting application logic - **Test deterministically**: Use mock implementations to create predictable test scenarios - **Compose flexibly**: Combine different implementations to create sophisticated behaviors - **Monitor comprehensively**: Inject observability concerns without modifying core logic ## The Bigger Picture Dependency injection in Xaibo isn't just a technical pattern, it's part of a broader architectural philosophy that prioritizes explicitness over convenience, flexibility over simplicity, and long-term maintainability over short-term ease of use. This philosophy recognizes that AI systems, like all complex software, benefit from careful architectural planning. By making dependencies explicit and manageable, Xaibo creates a foundation that can evolve with changing requirements and advancing technology. The dependency injection system works hand-in-hand with Xaibo's protocol-driven architecture and transparent proxy system to create a framework that's both powerful and maintainable. Each component focuses on its core responsibilities while the framework handles the complex work of wiring everything together. # Protocol-Driven Architecture At the heart of Xaibo lies a fundamental architectural decision: components communicate through protocols rather than concrete implementations. This choice shapes everything about how the framework operates and why it's particularly well-suited for building maintainable AI systems. ## The Problem with Direct Dependencies Traditional software often suffers from tight coupling, components that know too much about each other's internal workings. In AI systems, this problem is amplified. Consider a typical agent that needs to: - Generate responses using a language model - Store and retrieve memories - Execute tools - Handle different types of user input Without careful design, these components become entangled. The orchestrator might directly import and instantiate an OpenAI client, hardcoding assumptions about API keys, model names, and response formats. The memory system might be tightly coupled to a specific vector database. Tool execution might assume a particular function calling format. This tight coupling creates a cascade of problems: testing becomes difficult because you can't easily substitute mock components, switching providers requires rewriting core logic, and adding new capabilities means modifying existing code in unpredictable ways. ## Protocols as Contracts Xaibo solves this through protocols, formal contracts that define what a component can do without specifying how it does it. A protocol is essentially an interface that captures the essential behavior of a component while hiding its implementation details. Consider the [`LLMProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/llm.py). It defines a single essential operation: given a conversation, produce a response. It doesn't care whether that response comes from OpenAI's GPT-4, Anthropic's Claude, a local model, or even a simple rule-based system. The protocol captures the *what* while leaving the *how* completely open. ```python class LLMProtocol(Protocol): async def generate(self, messages: List[LLMMessage], options: Optional[LLMOptions] = None) -> LLMResponse: """Generate a response to a conversation.""" ... ``` This abstraction is powerful because it creates a stable interface that other components can depend on. An orchestrator that needs language model capabilities can declare its dependency on `LLMProtocol` without knowing or caring about the specific implementation. ## The Network Effect of Protocols The real power of protocol-driven architecture emerges when multiple protocols work together. Each protocol defines a clean boundary, and these boundaries compose naturally to create complex systems. The [`ToolProviderProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/tools.py) defines how tools are discovered and executed. The [`MemoryProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/memory.py) defines how information is stored and retrieved. The [`ResponseProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/response.py) defines how responses are delivered to users. An orchestrator module can declare dependencies on all these protocols without knowing anything about their implementations. This creates a powerful separation of concerns: the orchestrator focuses on coordination logic, while specialized modules handle the details of language generation, tool execution, and memory management. The protocol system also enables sophisticated aggregation patterns, where modules like [`ToolCollector`](../../../reference/modules/tools/#toolcollector) can implement a protocol while internally coordinating multiple providers of the same protocol. This creates unified interfaces that hide complexity while maintaining the flexibility of the underlying protocol system. ## Flexibility Through Abstraction This protocol-driven approach enables remarkable flexibility. Want to switch from OpenAI to Anthropic? Simply swap the LLM module, no changes needed to the orchestrator or any other component. Need to add a new type of memory system? Implement the `MemoryProtocol` and the rest of the system automatically gains access to it. The abstraction also enables sophisticated compositions. The [`LLMCombinator`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/combinator.py) module implements the `LLMProtocol` while internally coordinating multiple other LLM implementations. From the perspective of components that depend on it, it's just another LLM, but internally it might route different types of requests to different models, or combine responses from multiple models. ## Testing Through Protocols Protocols make testing dramatically easier. Instead of mocking complex external services, you can create simple implementations that fulfill the protocol contract. The [`MockLLM`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/mock.py) module provides predictable responses for testing, while still implementing the same `LLMProtocol` that production systems use. This isn't just convenient, it's transformative for testing AI systems. You can write deterministic tests that verify orchestration logic without the unpredictability of real language models. You can simulate error conditions, test edge cases, and verify behavior under controlled conditions. ## Evolution and Compatibility Protocols also provide a path for evolution. As AI capabilities advance and new patterns emerge, protocols can be extended while maintaining backward compatibility. New methods can be added to protocols, and existing implementations can gradually adopt them. The protocol system also enables gradual migration. You can introduce new implementations alongside existing ones, test them in isolation, and switch over when you're confident they work correctly. The protocol boundary provides a natural seam for this kind of evolution. ## The Cost of Abstraction Like any architectural decision, protocol-driven design involves trade-offs. The abstraction layer adds some complexity, you need to understand both the protocol interface and the specific implementations you're using. There's also a small performance overhead from the indirection, though in practice this is negligible compared to the cost of operations like language model inference. The bigger cost is conceptual: developers need to think in terms of interfaces and implementations rather than concrete objects. This requires a shift in mindset, but it's a shift that pays dividends as systems grow in complexity. ## Protocols in the Broader Ecosystem Xaibo's protocol-driven architecture reflects broader trends in software engineering. Microservices architectures use similar principles to create loosely coupled systems. Dependency injection frameworks provide similar benefits for managing component relationships. The Go programming language's interface system demonstrates how protocols can be both simple and powerful. In the context of AI systems, protocols are particularly valuable because the field is evolving so rapidly. Today's state-of-the-art model will be replaced by tomorrow's breakthrough. Today's preferred vector database will be superseded by next year's innovation. By building on stable protocol abstractions rather than specific implementations, Xaibo systems can evolve with the field rather than being locked into particular technologies. ## The Protocol Ecosystem Xaibo's core protocols form a coherent ecosystem that covers the essential aspects of AI agent behavior: - **Communication protocols** ([`TextMessageHandlerProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/message_handlers.py), [`ResponseProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/response.py)) handle interaction with users - **Cognitive protocols** ([`LLMProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/llm.py), [`MemoryProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/memory.py)) manage reasoning and knowledge - **Action protocols** ([`ToolProviderProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/tools.py)) enable interaction with external systems - **Coordination protocols** ([`ConversationHistoryProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/conversation.py)) manage state and flow This ecosystem provides a complete foundation for building AI agents while remaining open to extension and modification. New protocols can be added for emerging capabilities, and existing protocols can evolve to support new patterns. The protocol-driven architecture is more than just a technical choice, it's a philosophy that prioritizes flexibility, testability, and long-term maintainability over short-term convenience. It reflects the recognition that AI systems, like all complex software, benefit from careful architectural planning and clean abstractions. # How Transparent Proxies Enable Observability One of Xaibo's most distinctive features is its transparent proxy system, a "two-way mirror" that provides complete visibility into your agent's operations without requiring any changes to your code. This system represents a fundamental shift in how observability is approached in software systems, moving from explicit instrumentation to automatic, comprehensive monitoring. ## The Observability Challenge in AI Systems AI systems present unique observability challenges. Unlike traditional software where you can predict execution paths and identify key metrics in advance, AI agents are inherently non-deterministic. They make decisions based on complex reasoning processes, interact with external services in unpredictable ways, and often exhibit emergent behaviors that weren't explicitly programmed. Traditional approaches to observability require developers to manually instrument their code, adding logging statements, metrics collection, and tracing calls throughout the application. This approach has several problems: **Incomplete coverage**: Developers must remember to instrument every important operation, and they often miss edge cases or forget to update instrumentation when code changes. **Performance overhead**: Explicit instrumentation adds code that must be maintained and can impact performance. **Coupling concerns**: Business logic becomes mixed with observability concerns, making code harder to read and maintain. **Inconsistent data**: Different developers instrument code differently, leading to inconsistent observability data across the system. ## The Proxy Pattern Xaibo solves these problems using the proxy pattern, a design pattern where one object acts as a placeholder or surrogate for another object. In Xaibo's case, every component is automatically wrapped in a [`Proxy`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/exchange.py) that intercepts all method calls and provides observability without the component knowing it's being observed. When you request a module from Xaibo's exchange system, you don't get the module directly. Instead, you get a proxy that forwards all operations to the real module while capturing detailed information about every interaction: ```python # When you call this... response = await llm.generate(messages, options) # The proxy intercepts the call, captures timing and parameters, # forwards it to the real LLM, captures the response, # and emits events about the entire interaction ``` This interception is completely transparent to both the caller and the called component. The LLM module doesn't know it's being proxied, and the code calling it doesn't need to change. Yet every aspect of the interaction is captured and made available for analysis. ## Comprehensive Event Capture The proxy system captures far more information than traditional logging approaches. For every method call, Xaibo records: **Call details**: Method name, parameters, timestamp, and unique call identifier **Execution context**: Which module called which other module, creating a complete call graph **Timing information**: Precise timing for every operation, enabling performance analysis **Results and exceptions**: Complete response data and any exceptions that occurred **Relationship mapping**: How different components interact with each other over time This data is emitted as structured events through Xaibo's event system, where it can be consumed by various listeners for different purposes, debugging, performance monitoring, test case generation, or audit trails. ## The Two-Way Mirror Metaphor The "two-way mirror" metaphor captures an important aspect of Xaibo's proxy system. Like a two-way mirror, the proxy is transparent from one side (the components being observed don't know they're being watched) but provides complete visibility from the other side (observers can see everything that's happening). This transparency is crucial for several reasons: **No code changes required**: Existing components work unchanged when observability is added or removed. **Consistent instrumentation**: Every component gets the same level of observability automatically. **Zero performance impact when not needed**: Proxies can be disabled entirely for production deployments where observability isn't needed. **Complete coverage**: Nothing can hide from the proxy system, every interaction is captured. ## Method-Level Granularity The proxy system operates at the method level, which provides the right balance of detail and usability. It's fine-grained enough to capture the essential interactions between components, but not so detailed that it becomes overwhelming. Each method call gets wrapped in a [`MethodProxy`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/exchange.py) that handles the actual interception and event emission. This method-level approach means that complex operations are automatically broken down into their constituent parts, making it easy to understand how high-level behaviors emerge from low-level interactions. For example, when an orchestrator processes a user message, the proxy system automatically captures: - The initial message handling - LLM calls for reasoning - Tool executions - Memory retrievals and updates - Response generation Each of these operations is captured as separate events, but they're linked together through call identifiers and timing information, allowing you to reconstruct the complete flow of execution. ## Event-Driven Architecture The proxy system integrates seamlessly with Xaibo's event-driven architecture. Rather than writing observability data to logs or databases directly, proxies emit structured events that can be consumed by multiple listeners simultaneously. This approach provides tremendous flexibility: **Debug UI**: The development server includes a visual interface that consumes proxy events to show real-time execution flows and component interactions. **Test generation**: Events can be captured during manual testing and used to generate automated test cases. **Performance monitoring**: Timing events can be aggregated to identify performance bottlenecks. **Audit trails**: Complete interaction histories can be maintained for compliance or debugging purposes. **Custom analytics**: Application-specific event listeners can extract domain-specific insights from the interaction data. ## Zero-Overhead Observability One of the most elegant aspects of Xaibo's proxy system is that it can provide zero-overhead observability when not needed. If no event listeners are registered, the proxy system still intercepts calls but doesn't perform any expensive operations like serializing parameters or emitting events. This means you can deploy the same code to production with observability disabled for maximum performance, then enable it in development or debugging scenarios where the insights are more valuable than the performance cost. ## Comparison with Traditional Approaches Traditional observability approaches in software systems typically fall into several categories: **Logging**: Manual insertion of log statements throughout the code. Provides good detail but requires discipline and maintenance. **Metrics**: Collection of quantitative data about system behavior. Good for monitoring but lacks context about individual operations. **Tracing**: Tracking requests as they flow through distributed systems. Excellent for understanding system-wide behavior but requires explicit instrumentation. **Profiling**: Detailed analysis of code execution for performance optimization. Very detailed but typically used only during development. Xaibo's proxy system combines the benefits of all these approaches while avoiding their main drawbacks. It provides the detail of logging, the quantitative data of metrics, the flow tracking of tracing, and the comprehensive coverage of profiling, all without requiring explicit instrumentation. ## The Cost of Transparency Like any architectural decision, transparent proxies involve trade-offs. The proxy layer adds a small amount of overhead to every method call, though this is typically negligible compared to the cost of operations like LLM inference or tool execution. More significantly, the proxy system can generate large amounts of observability data, especially in complex agents with many component interactions. This data needs to be managed carefully to avoid overwhelming storage systems or analysis tools. There's also a conceptual cost: developers need to understand that they're working with proxies rather than direct objects. In practice, this is rarely an issue because the proxies are designed to be completely transparent, but it can occasionally matter for advanced use cases involving object identity or introspection. ## Observability as a First-Class Concern Xaibo's proxy system reflects a broader philosophy that observability should be a first-class concern in software architecture, not an afterthought. By building observability into the framework's core rather than treating it as an add-on, Xaibo ensures that every component benefits from comprehensive monitoring without additional effort from developers. This approach is particularly valuable in AI systems, where understanding what happened and why is often as important as the final result. The proxy system makes it possible to debug complex agent behaviors, optimize performance bottlenecks, and gain insights into how different components contribute to overall system behavior. ## Integration with Development Workflow The transparent proxy system integrates naturally with development workflows. During development, the rich observability data helps developers understand how their agents behave and identify issues quickly. The visual debug interface provides immediate feedback about component interactions and performance characteristics. During testing, the proxy system can capture complete interaction traces that serve as both documentation of expected behavior and regression tests for future changes. This automatic test case generation is particularly valuable for AI systems, where manual test case creation can be challenging due to the non-deterministic nature of the components. In production, the proxy system can be configured to capture only essential metrics, providing ongoing visibility into system health without the overhead of complete interaction logging. ## The Future of Observability Xaibo's transparent proxy system represents a vision of what observability could look like in future software systems. Rather than requiring developers to manually instrument their code, the framework automatically provides comprehensive visibility into system behavior. This approach becomes increasingly valuable as systems become more complex and distributed. AI agents, with their multiple interacting components and non-deterministic behaviors, are just one example of systems that benefit from this level of automatic observability. The proxy system also enables new possibilities for system analysis and optimization. With complete interaction data available, it becomes possible to apply machine learning techniques to understand system behavior, predict performance issues, or automatically optimize component interactions. By making observability transparent and automatic, Xaibo removes one of the traditional barriers to building maintainable, debuggable software systems. The result is a framework where understanding what your system is doing is as easy as building it in the first place. # Understanding Xaibo's Event System Xaibo's event system is the nervous system of the framework, a comprehensive observability layer that captures every interaction between components and makes it available for analysis, debugging, and optimization. Unlike traditional logging or monitoring approaches, Xaibo's event system is built into the framework's core architecture, providing automatic, comprehensive visibility into agent behavior. ## The Observability Gap in AI Systems AI systems present unique observability challenges that traditional software monitoring approaches struggle to address. When a language model generates an unexpected response, when a tool execution fails, or when an agent exhibits emergent behavior, understanding what happened requires more than just logs or metrics. Traditional observability approaches are designed for predictable, deterministic systems. They assume you know in advance what events are important, what metrics to collect, and what traces to follow. AI systems violate these assumptions, they're inherently non-deterministic, they exhibit complex emergent behaviors, and their most interesting events are often the ones you didn't anticipate. Xaibo's event system addresses this gap by capturing everything automatically. Instead of requiring developers to instrument their code with logging statements or metrics collection, the framework automatically emits detailed events for every component interaction. This comprehensive approach ensures that when something unexpected happens, you have the data needed to understand why. ## Event-Driven Architecture At its core, Xaibo's event system follows an event-driven architecture pattern. Components don't directly communicate with monitoring systems, instead, they emit events that can be consumed by multiple listeners simultaneously. This decoupling provides tremendous flexibility in how observability data is processed and used. The event system is built around the [`Event`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/models/events.py) model, which captures comprehensive information about each component interaction: - **Event identification**: Unique identifiers and timestamps for correlation - **Component context**: Which module called which method on which other module - **Execution details**: Parameters, return values, timing, and any exceptions - **Relationship mapping**: How different components interact over time This rich event model enables sophisticated analysis of agent behavior, from simple debugging to complex performance optimization and behavioral analysis. ## Automatic Event Generation The magic of Xaibo's event system lies in its automatic generation. Events are created transparently by the proxy system, every time one component calls a method on another component, the proxy automatically emits events capturing the interaction. This automatic generation has several important properties: **Comprehensive coverage**: Every component interaction is captured, not just the ones developers remember to instrument. **Consistent format**: All events follow the same structure, making them easier to analyze and correlate. **Zero maintenance overhead**: Events are generated automatically as code evolves, without requiring updates to instrumentation. **Performance awareness**: Event generation can be disabled entirely when observability isn't needed, eliminating any performance impact. The automatic generation also captures information that would be difficult or impossible to collect manually, such as precise timing information, complete parameter serialization, and detailed exception traces. ## Event Types and Lifecycle Xaibo's event system captures the complete lifecycle of component interactions through different event types: **CALL events**: Emitted when a method is invoked, capturing the parameters and context of the call. **RESULT events**: Emitted when a method completes successfully, capturing the return value and execution time. **EXCEPTION events**: Emitted when a method throws an exception, capturing the complete exception trace and context. This lifecycle approach provides complete visibility into both successful operations and failure modes. You can see not just what happened, but how long it took and what went wrong when things didn't work as expected. The event lifecycle also enables sophisticated analysis patterns. You can correlate CALL and RESULT events to understand performance characteristics, analyze EXCEPTION events to identify common failure modes, or track the flow of data through complex component interactions. ## Event Listeners and Processing The event system's power comes from its flexibility in processing events. Multiple event listeners can be registered to consume events simultaneously, each processing them for different purposes: **Debug visualization**: The development UI consumes events to provide real-time visualization of agent operations. **Performance monitoring**: Listeners can aggregate timing information to identify bottlenecks and optimization opportunities. **Test generation**: Events can be captured during manual testing and used to generate automated test cases. **Audit trails**: Complete interaction histories can be maintained for compliance or debugging purposes. **Custom analytics**: Application-specific listeners can extract domain-specific insights from the interaction data. Event listeners are registered with optional prefixes, allowing them to filter events based on component names, method names, or other criteria. This filtering capability ensures that listeners only receive the events they're interested in, improving performance and reducing noise. ## Correlation and Tracing One of the most powerful aspects of Xaibo's event system is its ability to correlate events across component boundaries. Each event includes correlation identifiers that link related operations together, enabling distributed tracing-style analysis of agent behavior. When an orchestrator processes a user message, it might call the LLM multiple times, execute several tools, and update memory. The event system automatically captures all these interactions and links them together through correlation identifiers. This enables you to reconstruct the complete flow of execution, understanding how high-level behaviors emerge from low-level component interactions. The correlation system also handles complex scenarios like parallel execution, where multiple operations might be happening simultaneously. Each operation gets its own correlation context, preventing confusion between concurrent activities. ## Real-Time and Historical Analysis The event system supports both real-time and historical analysis of agent behavior. Events can be consumed in real-time for immediate feedback and debugging, or stored for later analysis and optimization. Real-time event processing enables immediate visibility into agent operations. The debug UI shows component interactions as they happen, making it easy to understand what an agent is doing and identify issues quickly. This real-time feedback is particularly valuable during development and debugging. Historical event analysis enables deeper insights into agent behavior over time. By analyzing patterns in event data, you can identify performance trends, understand how agents behave in different scenarios, and optimize component interactions for better performance. ## Event Serialization and Storage Events are designed to be serializable, enabling them to be stored in databases, sent over networks, or processed by external systems. The event model includes careful handling of complex data types, ensuring that events can be faithfully reconstructed even when they contain large or complex objects. The serialization system also handles privacy and security concerns. Sensitive data can be filtered or redacted from events before they're stored or transmitted, ensuring that observability doesn't compromise security. Event storage can be configured based on specific needs. Development environments might store all events for comprehensive debugging, while production environments might store only summary information or events related to errors and performance issues. ## Performance Considerations While comprehensive event generation might seem expensive, Xaibo's event system is designed to minimize performance impact. Several design decisions contribute to this efficiency: **Lazy serialization**: Event data is only serialized when it's actually consumed by a listener, avoiding unnecessary work when events aren't being processed. **Configurable detail levels**: The amount of information captured in events can be configured based on specific needs, allowing you to balance observability with performance. **Efficient filtering**: Event listeners can filter events at the source, preventing unnecessary processing of events they don't need. **Optional generation**: Event generation can be disabled entirely in production environments where observability isn't needed. These performance optimizations ensure that the event system provides comprehensive observability when needed without imposing significant overhead when it's not. ## Integration with Development Workflow The event system integrates naturally with development workflows, providing immediate feedback about agent behavior and making it easier to understand complex systems. During development, the rich event data helps developers understand how their agents behave and identify issues quickly. The visual debug interface provides immediate feedback about component interactions, performance characteristics, and error conditions. During testing, the event system can capture complete interaction traces that serve as both documentation of expected behavior and regression tests for future changes. This automatic test case generation is particularly valuable for AI systems, where manual test case creation can be challenging due to the non-deterministic nature of the components. ## Comparison with Traditional Approaches Traditional observability approaches in software systems typically fall into several categories, each with their own strengths and limitations: **Application logs**: Provide detailed information about specific events but require manual instrumentation and can be inconsistent across different components. **Metrics and monitoring**: Provide quantitative data about system behavior but lack context about individual operations and their relationships. **Distributed tracing**: Tracks requests as they flow through systems but requires explicit instrumentation and may miss important interactions. **Profiling tools**: Provide detailed performance information but are typically used only during development and may not capture the full context of operations. Xaibo's event system combines the benefits of all these approaches while avoiding their main limitations. It provides the detail of application logs, the quantitative data of metrics, the flow tracking of distributed tracing, and the performance insights of profiling, all without requiring explicit instrumentation. ## Event System Extensibility The event system is designed to be extensible, supporting new event types and processing patterns as they emerge. Custom event listeners can be created to extract application-specific insights from the interaction data, and new event types can be added to capture domain-specific information. The extensibility also supports integration with external monitoring and analytics systems. Events can be forwarded to systems like Prometheus, Grafana, or custom analytics platforms, enabling integration with existing observability infrastructure. ## Privacy and Security Considerations The comprehensive nature of Xaibo's event system raises important privacy and security considerations. Events can contain sensitive information like user inputs, API keys, or proprietary data. The event system includes several mechanisms to address these concerns: **Data filtering**: Sensitive fields can be filtered out of events before they're stored or transmitted. **Encryption**: Events can be encrypted when stored or transmitted over networks. **Access controls**: Event data can be protected with appropriate access controls to ensure only authorized users can access it. **Retention policies**: Event data can be automatically purged after specified time periods to minimize privacy exposure. These privacy and security features ensure that the benefits of comprehensive observability don't come at the cost of data protection. ## The Future of AI Observability Xaibo's event system represents a vision of what observability could look like in future AI systems. As AI systems become more complex and autonomous, understanding their behavior becomes increasingly important for debugging, optimization, and safety. The comprehensive event data captured by Xaibo's system enables new possibilities for AI system analysis. Machine learning techniques can be applied to event data to identify patterns, predict failures, or automatically optimize system behavior. The rich interaction data can also support more sophisticated debugging tools and development environments. By making observability automatic and comprehensive, Xaibo removes one of the traditional barriers to building maintainable, debuggable AI systems. The result is a framework where understanding what your system is doing is as natural as building it in the first place. ## The Philosophical Dimension Xaibo's event system reflects a broader philosophy about software observability. Instead of treating observability as an afterthought or add-on, the framework makes it a first-class concern that's built into the architecture from the ground up. This approach recognizes that in complex systems, and AI systems are inherently complex, understanding what happened is often as important as making it happen in the first place. The event system provides the foundation for this understanding, enabling developers to build systems that are not just functional, but comprehensible and maintainable over time. # How the Exchange System Works The exchange system is Xaibo's dependency injection container, the sophisticated machinery that takes a collection of module definitions and automatically wires them together into a functioning agent. Understanding how this system works illuminates not just the technical mechanics of Xaibo, but the architectural philosophy that makes the framework so flexible and maintainable. ## The Wiring Problem Imagine you're assembling a complex electronic device. You have dozens of components, each with specific input and output requirements. Some components need power, others need data signals, and still others need control inputs. The challenge isn't just connecting the components, it's figuring out the correct order to connect them, ensuring that each component gets what it needs when it needs it. This is exactly the problem that Xaibo's exchange system solves, but for software components instead of electronic ones. Each module has dependencies, other modules it needs to function properly. The exchange system analyzes these dependencies and automatically creates the necessary connections in the correct order. ## Dependency Analysis The exchange system begins by analyzing the constructor signatures of all modules in the agent configuration. When a module declares a parameter like `llm: LLMProtocol`, the exchange system understands that this module needs something that implements the `LLMProtocol` interface. This analysis happens automatically through Python's type annotation system. The exchange system uses introspection to examine each module's `__init__` method and extract the type information for each parameter. This approach means that dependencies are declared in the natural place, the constructor, rather than in separate configuration files or decorators. The dependency analysis also handles more complex scenarios. If a module declares a parameter like `tools: list[ToolsProtocol]`, the exchange system understands that this module needs a list of implementations, not just a single one. This enables modules like [`ToolCollector`](../../../reference/modules/tools/#toolcollector) that aggregate multiple tool providers into a single interface, allowing orchestrators to access tools from different sources through one unified connection point. ## Automatic Resolution Once the exchange system understands what each module needs, it attempts to automatically resolve these dependencies by matching them with available providers. If a module needs an `LLMProtocol` and only one module in the configuration provides that protocol, the connection is made automatically. This automatic resolution eliminates much of the boilerplate configuration that would otherwise be required. In simple cases, you can define your modules and let the exchange system figure out how to connect them. This reduces configuration complexity and makes it easier to get started with Xaibo. The automatic resolution also handles transitive dependencies. If module A depends on module B, and module B depends on module C, the exchange system ensures that C is instantiated before B, and B is instantiated before A. This dependency ordering is computed automatically using a topological sort algorithm. ## Explicit Configuration While automatic resolution works well for simple cases, complex systems often need explicit control over how dependencies are resolved. The exchange configuration provides this control through explicit binding specifications: ```yaml exchange: - module: orchestrator protocol: LLMProtocol provider: llm ``` This explicit configuration overrides the automatic matching, allowing you to specify exactly which implementations should be used for which dependencies. This is particularly useful when you have multiple implementations of the same protocol and need to control which one goes where. The explicit configuration also supports more sophisticated scenarios, like providing different implementations to different modules, or creating complex dependency graphs that wouldn't be resolvable automatically. ## Tool Aggregation Patterns The exchange system's handling of list dependencies enables sophisticated aggregation patterns, with [`ToolCollector`](../../../reference/modules/tools/#toolcollector) serving as the canonical example. This pattern demonstrates how the exchange system can compose multiple providers into unified interfaces, creating emergent capabilities that exceed the sum of their parts. When the exchange system encounters a module like `ToolCollector` that declares `tools: list[ToolProviderProtocol]`, it automatically collects all available tool providers and injects them as a list. The `ToolCollector` then creates a unified interface that aggregates tools from all providers, maintaining an internal cache that maps tool names to their originating providers. This aggregation pattern solves several architectural challenges simultaneously. It eliminates the need for orchestrators to manage multiple tool provider connections, provides a single point of tool discovery across the entire system, and enables dynamic tool ecosystems where providers can be added or removed without affecting consuming modules. This exemplifies the [modular design philosophy](../../design/modularity/#the-network-effect-of-modularity) where individual components create exponential value through composition. The exchange system's support for this pattern extends beyond simple list injection. The explicit configuration syntax allows fine-grained control over which providers are aggregated: ```yaml exchange: - module: tool-collector protocol: ToolProviderProtocol provider: [python-tools, mcp-tools, custom-tools] ``` This configuration creates a deliberate composition where the tool collector aggregates only the specified providers, enabling different tool collections for different parts of the system. The exchange system handles the complexity of ensuring all specified providers are available before instantiating the collector. The tool aggregation pattern also demonstrates the exchange system's role in enabling architectural evolution. As new tool providers are developed, they can be seamlessly integrated into existing systems by adding them to the collector's provider list. The consuming modules remain unchanged, benefiting from new capabilities without requiring modification. ## The Exchange as a Registry The [`Exchange`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/exchange.py) class acts as both a dependency injection container and a service registry. Once modules are instantiated, they're stored in the exchange and can be retrieved by other parts of the system. This registry function is important for several reasons. It ensures that modules are singletons within an agent, each module is instantiated exactly once and reused whenever it's needed. It also provides a central place to manage module lifecycles and handle cleanup when an agent is shut down. The exchange also handles the proxy wrapping that enables Xaibo's observability features. When a module is retrieved from the exchange, it's automatically wrapped in a proxy that captures method calls and emits events. This happens transparently, neither the requesting module nor the provided module needs to know about the proxy layer. ## Instantiation Order One of the most complex aspects of the exchange system is determining the correct order to instantiate modules. This isn't just a matter of convenience, it's essential for correctness. A module can't be instantiated until all of its dependencies are available. The exchange system solves this using a topological sort algorithm that analyzes the dependency graph and determines a valid instantiation order. This algorithm handles complex scenarios like modules with multiple dependencies, optional dependencies, and even circular dependencies (which are detected and reported as errors). The instantiation order computation also considers the specific types of dependencies. Simple dependencies must be resolved before the dependent module is instantiated, but list dependencies can be resolved incrementally as providers become available. ## Handling Ambiguity Sometimes the automatic resolution process encounters ambiguity, multiple modules provide the same protocol, or a module's dependencies can't be uniquely resolved. The exchange system handles these situations gracefully by providing clear error messages that explain what went wrong and suggest possible solutions. For example, if two modules both provide `LLMProtocol` and a third module depends on it, the exchange system will report that the dependency is ambiguous and suggest using explicit configuration to resolve it. This approach prevents silent failures and makes it easier to debug configuration issues. The exchange system also provides tools for introspecting the dependency graph, making it easier to understand how modules are connected and identify potential issues before they cause runtime problems. ## Override and Testing Support The exchange system includes sophisticated support for overriding dependencies, which is particularly important for testing. The [`ConfigOverrides`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/config.py) mechanism allows you to replace specific modules with alternative implementations without modifying the main configuration. This override capability is what makes Xaibo's testing story so compelling. You can create an exchange with mock implementations of external services, ensuring that tests are fast, reliable, and isolated from external dependencies. The override system handles all the complexity of rewiring dependencies while maintaining the same interface that production code expects. This demonstrates how [dependency injection architecture](../../architecture/dependency-injection/#testing-through-injection) enables superior testing strategies through explicit dependency management. The override system also supports partial overrides, where only some dependencies are replaced while others use the standard configuration. This enables sophisticated testing scenarios where you might want to test real LLM integration while mocking tool execution, or vice versa. ## Performance Considerations The exchange system is designed to minimize runtime overhead while providing maximum flexibility. The dependency analysis and instantiation ordering happen once, during agent initialization, rather than on every method call. Once modules are instantiated and wired together, the exchange system adds minimal overhead to normal operation. The proxy wrapping does add some overhead to method calls, but this overhead is configurable. In production deployments where observability isn't needed, the proxy system can be disabled entirely, eliminating any performance impact. The exchange system also supports lazy instantiation, where modules are only created when they're actually needed. This can improve startup time for complex agents with many modules, especially when some modules are only used in specific scenarios. ## Error Handling and Diagnostics The exchange system provides comprehensive error handling and diagnostic information. When something goes wrong during dependency resolution or module instantiation, the system provides detailed error messages that explain what happened and suggest possible fixes. The diagnostic information includes the complete dependency graph, showing which modules depend on which others and how those dependencies are resolved. This information is invaluable for debugging complex configuration issues or understanding how a particular agent is structured. The exchange system also validates configurations before attempting instantiation, catching many common errors early in the process. This validation includes checking for circular dependencies, missing providers, and type mismatches between dependencies and providers. ## Comparison with Other Approaches Different frameworks handle dependency injection in different ways. Some use constructor injection like Xaibo, while others use property injection, method injection, or service locator patterns. Each approach has trade-offs in terms of explicitness, flexibility, and ease of use. Xaibo's approach prioritizes explicitness and type safety. Dependencies are declared in constructor signatures where they're naturally visible, and the type system helps catch configuration errors early. This approach requires more upfront thinking about dependencies, but it pays dividends in terms of maintainability and debuggability. The automatic resolution capability sets Xaibo apart from many other dependency injection frameworks. While explicit configuration is always available when needed, the automatic resolution eliminates boilerplate in common cases and makes it easier to get started with the framework. ## The Exchange as an Architectural Pattern The exchange system embodies several important architectural patterns: **Inversion of Control**: Instead of modules controlling their own dependencies, the exchange controls what dependencies each module receives. **Dependency Injection**: Dependencies are provided to modules rather than being created or located by the modules themselves. **Service Locator**: The exchange acts as a central registry where modules can be located by their identifiers. **Factory Pattern**: The exchange handles the complex logic of creating and configuring modules. These patterns work together to create a system that's both powerful and easy to use. The exchange handles the complexity of dependency management while providing a simple, declarative interface for configuration. ## Evolution and Extensibility The exchange system is designed to evolve with the framework and support new patterns as they emerge. The plugin architecture allows new types of dependencies and resolution strategies to be added without modifying the core exchange logic. The exchange system also supports versioning and migration strategies for configurations. As protocols evolve and new module types are added, the exchange system can help migrate existing configurations to new patterns while maintaining backward compatibility. ## The Bigger Picture The exchange system is more than just a technical component, it's the embodiment of Xaibo's architectural philosophy. By making dependencies explicit and manageable, the exchange system enables the modularity, testability, and flexibility that make Xaibo systems maintainable over time. The exchange system also reflects the recognition that complex software systems benefit from careful architectural planning. Rather than leaving dependency management as an afterthought, Xaibo makes it a first-class concern with sophisticated tooling and clear patterns. Understanding the exchange system helps you understand not just how Xaibo works, but why it works the way it does. The system represents a careful balance between automation and control, between convenience and explicitness, and between simplicity and power. This balance is what makes Xaibo both approachable for newcomers and powerful for complex use cases. # Understanding the Relationship Between Modules and Protocols The distinction between modules and protocols is fundamental to understanding how Xaibo works, yet it's often a source of confusion for newcomers to the framework. This relationship, between what something *is* (a module) and what it *can do* (a protocol), shapes every aspect of how Xaibo systems are designed and composed. ## The Interface vs Implementation Distinction In software engineering, there's a crucial distinction between interface and implementation. An interface defines *what* operations are available, while an implementation defines *how* those operations are carried out. This distinction is the foundation of modular design, but it's often blurred in practice. Xaibo makes this distinction explicit and central to its architecture. Protocols define interfaces, they specify what operations a component must support without dictating how those operations are implemented. Modules provide implementations, they contain the actual code that carries out the operations defined by protocols. Consider the [`LLMProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/llm.py). It defines a single essential operation: given a conversation, generate a response. This protocol doesn't care about API keys, model names, or inference techniques, it only cares about the fundamental capability of generating responses to conversations. Multiple modules can implement this same protocol: [`OpenAILLM`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/openai.py), [`AnthropicLLM`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/anthropic.py), [`MockLLM`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/mock.py), and others. Each module provides a different implementation of the same interface, enabling them to be used interchangeably. ## The Power of Substitutability This separation enables one of the most powerful properties in software design: substitutability. Any module that implements a protocol can be substituted for any other module that implements the same protocol, without affecting the rest of the system. This substitutability is what makes Xaibo systems so flexible. Want to switch from OpenAI to Anthropic? Both modules implement `LLMProtocol`, so you can swap them without changing any other component. Need to test your orchestrator logic? Use `MockLLM` instead of a real language model, the orchestrator doesn't know the difference. The substitutability extends beyond simple swapping. You can create sophisticated compositions by implementing protocols in creative ways. The [`LLMCombinator`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/combinator.py) module implements `LLMProtocol` while internally coordinating multiple other LLM implementations. From the outside, it looks like a single LLM, but internally it might route different types of requests to different models. ## Multiple Protocols, Single Module The relationship between modules and protocols isn't one-to-one. A single module can implement multiple protocols, providing different capabilities to different parts of the system. Consider a hypothetical `SmartMemory` module that implements both [`MemoryProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/memory.py) and [`ToolProviderProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/tools.py). As a memory provider, it stores and retrieves information. As a tool provider, it offers tools for querying and manipulating that same information. The same underlying implementation serves both roles, but through different protocol interfaces. This multi-protocol approach enables rich, cohesive modules that provide related capabilities through well-defined interfaces. It also enables more efficient implementations, the `SmartMemory` module can share data structures and optimizations between its memory and tool capabilities. ## Protocol Composition and Layering Protocols can build on each other to create layered abstractions. Some protocols are fundamental, they define basic capabilities that many modules need. Others are specialized, they define specific behaviors for particular use cases. The [`ResponseProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/response.py) is fundamental, it defines how responses are delivered to users. The [`TextMessageHandlerProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/message_handlers.py) is more specialized, it defines how text messages are processed. This layering enables modules to focus on their core competencies while relying on other modules for supporting capabilities. An orchestrator module might implement `TextMessageHandlerProtocol` for processing user messages while depending on other modules that implement `LLMProtocol`, `ToolProviderProtocol`, and `ResponseProtocol`. ## The Evolution Problem One of the challenges in any modular system is evolution, how do you add new capabilities without breaking existing components? The protocol-module distinction provides a elegant solution to this problem. Protocols can evolve by adding new optional methods while maintaining backward compatibility. Existing modules continue to work with the old interface, while new modules can take advantage of enhanced capabilities. The framework can detect which methods a module supports and adapt accordingly. Modules can also evolve independently of the protocols they implement. A new, more efficient implementation of `LLMProtocol` can be dropped into an existing system without requiring changes to any other component. The protocol interface provides a stable contract that enables this kind of independent evolution. ## Testing Through Protocol Boundaries The protocol-module distinction is particularly powerful for testing. Instead of trying to mock complex external services, you can create simple test modules that implement the same protocols as production modules. A test module implementing `LLMProtocol` might return predefined responses, simulate various error conditions, or introduce controlled delays. Because it implements the same protocol as production LLM modules, it can be used as a drop-in replacement for testing purposes. This approach makes tests more reliable and faster. Instead of depending on external services with their own availability and rate limiting concerns, tests can use predictable local implementations. The tests focus on the logic being tested rather than the complexities of external integrations. ## The Cognitive Model Understanding the module-protocol relationship requires a shift in how you think about software components. Instead of thinking about concrete objects with specific capabilities, you need to think about roles and responsibilities. A protocol defines a role, "something that can generate language model responses" or "something that can execute tools." A module fills that role by providing a concrete implementation. The same module might fill multiple roles, and multiple modules might compete to fill the same role. This role-based thinking is particularly important when designing new components. Instead of asking "what does this module do?", you ask "what roles does this module fill?" and "what roles does this module depend on?" This perspective leads to more modular, flexible designs. ## Protocol Discovery and Introspection Xaibo's exchange system automatically discovers which protocols a module implements by examining its type annotations and method signatures. This automatic discovery means that modules don't need to explicitly declare which protocols they implement, the framework figures it out automatically. This approach reduces boilerplate and makes it easier to create new modules. You simply implement the methods defined by the protocols you want to support, and the framework handles the rest. It also makes the system more robust, there's no way for a module to claim to implement a protocol without actually providing the required methods. ## The Network Effect The real power of the protocol-module distinction emerges when you have many modules implementing overlapping sets of protocols. This creates a rich ecosystem where components can be combined in unexpected ways. Consider a system with multiple LLM modules, multiple tool modules, and multiple memory modules. The number of possible combinations grows exponentially, but because they all interact through well-defined protocol interfaces, any combination that makes logical sense will work correctly. This network effect is particularly valuable in AI systems, where new capabilities and providers are constantly emerging. The protocol-based architecture means that new modules can be integrated into existing systems without requiring changes to other components. ## Design Patterns and Best Practices The protocol-module relationship enables several important design patterns: **Adapter Pattern**: Create modules that implement one protocol while delegating to modules that implement different protocols. This enables integration between incompatible interfaces. **Decorator Pattern**: Create modules that implement a protocol while adding behavior to other modules that implement the same protocol. This enables cross-cutting concerns like caching, logging, or rate limiting. **Strategy Pattern**: Use different modules implementing the same protocol to provide different algorithms or behaviors. This enables runtime selection of implementation strategies. **Composite Pattern**: Create modules that implement a protocol while coordinating multiple other modules that implement the same protocol. This enables sophisticated orchestration and load balancing. ## The Philosophical Dimension The protocol-module distinction reflects a deeper philosophical approach to software design. It prioritizes flexibility over convenience, explicitness over implicitness, and long-term maintainability over short-term simplicity. This approach recognizes that software systems evolve over time, and that the decisions you make early in a project's lifecycle will constrain your options later. By building on stable protocol abstractions rather than concrete implementations, Xaibo systems can evolve gracefully as requirements change and new technologies emerge. The distinction also reflects the recognition that AI systems are inherently complex and that this complexity is best managed through careful architectural planning rather than ad-hoc solutions. The protocol-module relationship provides a principled way to manage this complexity while maintaining flexibility for future evolution. ## Practical Implications Understanding the protocol-module relationship has practical implications for how you design and build Xaibo systems: **Think in terms of capabilities**: When designing a new component, focus on what capabilities it provides (protocols) rather than how it provides them (implementation details). **Design for substitutability**: Ensure that your modules can be easily replaced by alternative implementations of the same protocols. **Minimize protocol dependencies**: Depend on the smallest set of protocols necessary to accomplish your goals. This makes your modules more reusable and easier to test. **Consider protocol evolution**: Design protocols that can evolve over time without breaking existing implementations. **Leverage composition**: Use multiple simple modules implementing focused protocols rather than single complex modules implementing many protocols. The protocol-module relationship is more than just a technical pattern, it's a way of thinking about software architecture that prioritizes flexibility, testability, and long-term maintainability. By understanding this relationship, you can build Xaibo systems that are not just functional, but elegant and sustainable. # Understanding Xaibo's Extensibility Model Extensibility, the ability to add new capabilities without modifying existing code, is one of the most critical characteristics of any framework intended for long-term use. In the rapidly evolving AI landscape, where new models, techniques, and integration patterns emerge regularly, extensibility isn't just a nice-to-have feature, it's essential for survival. Xaibo's extensibility model is designed to accommodate not just today's requirements, but tomorrow's innovations that we can't yet imagine. ## The Extensibility Challenge Building extensible systems is fundamentally about managing the tension between stability and change. You need stable interfaces that existing code can depend on, but you also need the flexibility to add new capabilities as requirements evolve. This tension is particularly acute in AI systems, where the underlying technology is advancing rapidly and new patterns emerge frequently. Traditional approaches to extensibility often fall into several problematic patterns: **Inheritance-Based Extension**: Using class hierarchies to enable customization. This approach quickly becomes unwieldy as the number of variations grows, and it creates tight coupling between base classes and extensions. **Configuration-Driven Extension**: Using configuration files or parameters to control behavior. This works for simple variations but becomes complex and error-prone as the number of options grows. **Plugin Architectures**: Defining extension points where new code can be plugged in. This works well but requires careful design of the extension interfaces and can be limiting when the core assumptions need to change. **Fork-and-Modify**: Simply copying the codebase and modifying it for specific needs. This provides maximum flexibility but eliminates the benefits of shared development and makes it difficult to incorporate upstream improvements. ## Xaibo's Protocol-Based Extensibility Xaibo's approach to extensibility is built around its protocol-driven architecture. Instead of defining specific extension points or inheritance hierarchies, the framework defines protocols that capture essential behaviors. New capabilities are added by implementing these protocols, and existing capabilities can be extended by creating new implementations that build on or compose existing ones. This protocol-based approach provides several important extensibility characteristics: **Open Extension**: New implementations can be added without modifying existing code. The protocol interfaces provide stable contracts that new implementations can fulfill. **Compositional Extension**: Complex behaviors can be built by composing simpler implementations. The [`LLMCombinator`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/combinator.py) demonstrates this pattern by implementing `LLMProtocol` while coordinating multiple other LLM implementations. **Substitutional Extension**: New implementations can replace existing ones without affecting other components. This enables gradual migration to new technologies or approaches. **Additive Extension**: New protocols can be defined to capture emerging patterns, and existing modules can be extended to implement these new protocols. ## Extension Through Implementation The most straightforward way to extend Xaibo is by creating new implementations of existing protocols. This pattern enables adding support for new providers, new algorithms, or new integration patterns without modifying the framework core. Consider adding support for a new language model provider. You implement the [`LLMProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/llm.py) interface, handle the provider-specific API details, and the new implementation can be used anywhere that LLM capabilities are needed. The rest of the system doesn't need to know or care about the specific provider, it just uses the standard protocol interface. This implementation-based extension pattern works for all of Xaibo's core protocols: - **LLM Extensions**: New model providers, local models, specialized models for specific tasks - **Memory Extensions**: New vector databases, different embedding approaches, specialized memory architectures - **Tool Extensions**: New tool formats, different execution environments, specialized tool categories - **Response Extensions**: New output formats, different delivery mechanisms, specialized response processing ## Extension Through Composition Xaibo's architecture enables sophisticated extension patterns through composition, building complex behaviors by combining simpler components. This compositional approach enables creating new capabilities that leverage existing implementations while adding new behaviors. The composition pattern is particularly powerful for creating specialized behaviors: **Caching Layers**: Implement a protocol while adding caching behavior to an underlying implementation. This enables performance optimization without modifying existing components. **Retry Logic**: Wrap existing implementations with retry behavior to handle transient failures. This enables reliability improvements without changing core logic. **Load Balancing**: Distribute requests across multiple implementations of the same protocol. This enables scalability improvements without modifying individual components. **Monitoring and Instrumentation**: Add observability concerns to existing implementations without modifying their core behavior. These compositional patterns enable building sophisticated systems from simple, focused components while maintaining clear separation of concerns. ## Protocol Evolution One of the most challenging aspects of extensibility is evolving interfaces over time. As new requirements emerge, protocols need to be extended to support new capabilities while maintaining backward compatibility with existing implementations. Xaibo's protocol evolution strategy is based on several principles: **Additive Changes**: New methods can be added to protocols as optional capabilities. Existing implementations continue to work, while new implementations can take advantage of enhanced capabilities. **Default Implementations**: Protocol extensions can provide default implementations for new methods, enabling existing implementations to gain new capabilities automatically. **Capability Detection**: The framework can detect which methods an implementation supports and adapt behavior accordingly. This enables graceful degradation when advanced capabilities aren't available. **Versioning Support**: Protocols can be versioned to enable explicit compatibility management when breaking changes are necessary. This evolution strategy enables protocols to grow and adapt over time while maintaining stability for existing implementations. ## Custom Protocol Definition While Xaibo provides protocols for common AI system patterns, the framework also supports defining custom protocols for domain-specific requirements. This capability enables extending the framework's conceptual model to support new patterns that emerge in specific applications or domains. Custom protocol definition follows the same patterns as built-in protocols: **Interface Definition**: Define the methods and behaviors that implementations must provide. **Type Annotations**: Use Python's type system to specify parameter and return types for better tooling support and runtime validation. **Documentation**: Provide clear documentation of the protocol's purpose, usage patterns, and implementation requirements. **Reference Implementation**: Create at least one implementation to validate the protocol design and provide an example for other implementers. Custom protocols integrate seamlessly with Xaibo's dependency injection and event systems, providing the same benefits as built-in protocols. ## Extension Discovery and Registration Xaibo's extension model includes mechanisms for discovering and registering new implementations without requiring changes to the framework core. This enables creating extension packages that can be installed and used without modifying existing code. The discovery mechanism works through Python's module system and type annotations. When the framework encounters a new module that implements a known protocol, it automatically recognizes the implementation and makes it available for use. This automatic discovery eliminates the need for explicit registration while maintaining type safety. Extension packages can also include configuration templates, documentation, and examples that help users understand how to use the new capabilities. This packaging approach enables creating rich extension ecosystems around the framework. ## Backward Compatibility Maintaining backward compatibility is crucial for extensibility, extensions should continue to work as the framework evolves. Xaibo's approach to backward compatibility is based on several strategies: **Stable Protocol Interfaces**: Core protocols are designed to be stable over time, with changes made through additive evolution rather than breaking changes. **Deprecation Policies**: When changes are necessary, deprecated features are maintained for several versions with clear migration paths to new approaches. **Compatibility Testing**: The framework includes comprehensive tests that verify backward compatibility across versions. **Documentation**: Clear documentation of compatibility policies and migration strategies helps extension developers plan for changes. This commitment to backward compatibility enables building extension ecosystems that can evolve over time without constantly breaking existing implementations. ## Performance Considerations Extensibility often comes with performance trade-offs, the flexibility that enables extension can also introduce overhead. Xaibo's extensibility model is designed to minimize these trade-offs through several strategies: **Protocol Optimization**: Protocol interfaces are designed to minimize overhead while providing necessary flexibility. **Lazy Loading**: Extensions are only loaded when they're actually used, reducing startup overhead for systems that don't use all available capabilities. **Caching**: The framework caches protocol resolution and other expensive operations to minimize runtime overhead. **Optional Features**: Advanced extensibility features can be disabled in production environments where maximum performance is more important than flexibility. These optimizations ensure that extensibility doesn't come at the cost of performance in production deployments. ## Extension Ecosystem Patterns Xaibo's extensibility model enables several important ecosystem patterns that benefit both framework developers and users: **Community Contributions**: The protocol-based approach makes it easy for community members to contribute new implementations without requiring deep knowledge of the framework internals. **Vendor Integrations**: Service providers can create official implementations of Xaibo protocols, enabling seamless integration with their services. **Domain-Specific Extensions**: Specialized implementations can be created for specific domains or use cases, enabling the framework to serve diverse communities. **Experimental Features**: New ideas can be prototyped as extensions before being considered for inclusion in the framework core. These ecosystem patterns create a virtuous cycle where the framework becomes more valuable as more extensions are created, which in turn encourages more extension development. ## Testing Extension Points Xaibo's testing infrastructure is designed to support extension testing as well as core framework testing. Extension developers can use the same testing patterns and utilities that the framework itself uses: **Mock Implementations**: Create test implementations of protocols for testing extension logic in isolation. **Integration Testing**: Test extensions in combination with real framework components to verify correct integration. **Compatibility Testing**: Verify that extensions work correctly with different versions of the framework and other extensions. **Performance Testing**: Measure the performance impact of extensions and optimize accordingly. This comprehensive testing support ensures that extensions can be developed with the same quality standards as the framework core. ## Documentation and Examples Effective extensibility requires comprehensive documentation and examples that help developers understand how to create extensions. Xaibo's documentation strategy includes: **Protocol Documentation**: Clear specification of what each protocol does and how implementations should behave. **Implementation Guides**: Step-by-step guides for creating new implementations of existing protocols. **Extension Examples**: Complete examples that demonstrate common extension patterns. **Best Practices**: Guidelines for creating high-quality, maintainable extensions. This documentation approach ensures that the extensibility capabilities are accessible to developers with varying levels of experience with the framework. ## The Future of Extensibility Xaibo's extensibility model is designed to evolve with the framework and the broader AI ecosystem. Several trends are likely to influence how extensibility develops: **Standardization**: As AI system patterns mature, there may be opportunities to standardize protocols across frameworks, enabling greater interoperability. **Automation**: Tools for automatically generating extension scaffolding, testing, and documentation may reduce the effort required to create high-quality extensions. **Marketplace Models**: Extension distribution and discovery mechanisms may evolve to support marketplace-style ecosystems. **Cross-Language Support**: Extension mechanisms may be extended to support implementations in languages other than Python. By building extensibility into the framework's core architecture rather than treating it as an add-on feature, Xaibo positions itself to evolve with these trends while maintaining stability for existing users. ## The Philosophy of Extensibility Xaibo's approach to extensibility reflects a broader philosophy about software development that prioritizes long-term thinking over short-term convenience. The framework recognizes that requirements will change, new technologies will emerge, and today's best practices will be superseded by tomorrow's innovations. Rather than trying to predict these changes, the framework provides mechanisms for adapting to them. The protocol-based architecture, the dependency injection system, and the comprehensive testing support all work together to create a foundation that can evolve gracefully over time. This philosophy of extensibility isn't just about technical capabilities, it's about creating sustainable software ecosystems that can grow and adapt while maintaining stability and reliability. By making extensibility a first-class concern, Xaibo enables not just today's AI applications, but the AI applications of the future that we can't yet imagine. # Why Xaibo Chose Modularity Over Monolithic Design The decision to build Xaibo as a modular framework rather than a monolithic system represents a fundamental philosophical choice about how complex software should be structured. This choice affects every aspect of the framework, from how components are designed to how systems evolve over time. Understanding why modularity was chosen, and what alternatives were rejected, illuminates the deeper principles that guide Xaibo's architecture. ## The Monolithic Alternative To understand why Xaibo chose modularity, it's helpful to consider what a monolithic AI agent framework might look like. Such a framework would provide a single, integrated solution that handles all aspects of agent behavior: language model integration, tool execution, memory management, and user interaction would all be built into a unified system. This monolithic approach has some appealing characteristics. It's conceptually simpler, there's one system to understand rather than many interacting components. It can be more efficient, direct function calls are faster than protocol-mediated interactions. It's easier to optimize, the entire system can be tuned as a single unit. Many successful AI frameworks have taken this approach. They provide comprehensive solutions that work well out of the box, with minimal configuration required. For simple use cases or rapid prototyping, this can be exactly what's needed. ## The Limits of Monoliths However, monolithic systems have fundamental limitations that become apparent as requirements grow in complexity. These limitations aren't just technical, they're architectural and organizational. **Coupling and Dependencies**: In a monolithic system, components become tightly coupled over time. The language model integration might depend on specific memory formats, the tool execution system might assume particular response structures, and the user interface might be tied to specific orchestration patterns. These dependencies make it difficult to change any part of the system without affecting others. **Technology Lock-in**: Monolithic systems tend to lock you into specific technology choices. If the framework assumes you're using OpenAI's models, switching to Anthropic becomes a major undertaking. If it's built around a particular vector database, migrating to a different one requires significant rework. **Testing Complexity**: Testing monolithic systems is inherently difficult because you can't easily isolate individual components. Testing the orchestration logic requires setting up the entire system, including external dependencies like language models and databases. This makes tests slow, brittle, and expensive to run. **Evolution Challenges**: As AI technology evolves rapidly, monolithic systems struggle to keep up. Adding support for new model providers, new tool formats, or new interaction patterns requires changes throughout the system. The risk of breaking existing functionality grows with each change. ## The Modular Philosophy Xaibo's modular approach addresses these limitations by decomposing the system into independent, interchangeable components. Each module has a focused responsibility and interacts with other modules through well-defined protocols. This decomposition enables several important properties: **Independent Evolution**: Modules can evolve independently as long as they maintain their protocol contracts. A new LLM provider can be added without affecting memory management. A more efficient vector index can be swapped in without changing tool execution logic. **Technology Flexibility**: The modular approach enables mixing and matching technologies. You can use OpenAI for language generation, local embeddings for memory, and cloud services for tool execution, all within the same agent. **Focused Testing**: Each module can be tested in isolation using mock implementations of its dependencies. This makes tests faster, more reliable, and easier to understand. You can test orchestration logic without real language models, or memory management without external databases. **Compositional Complexity**: Complex behaviors emerge from the composition of simple modules rather than being built into monolithic components. This compositional approach makes it easier to understand, debug, and modify system behavior. ## The Cost of Modularity Modularity isn't free, it comes with costs that must be weighed against its benefits. Understanding these costs helps explain why modularity isn't always the right choice, and why Xaibo's particular approach to modularity is designed to minimize these costs. **Conceptual Overhead**: Modular systems require understanding multiple components and their interactions rather than a single unified system. Developers need to think about protocols, dependencies, and composition patterns. This conceptual overhead can be significant, especially for newcomers to the framework. **Performance Implications**: Protocol-mediated interactions are slower than direct function calls. The proxy system adds overhead to method calls. Dependency injection requires runtime resolution of component relationships. While these overheads are typically small compared to the cost of operations like language model inference, they're not zero. **Configuration Complexity**: Modular systems require more configuration to specify how components should be connected. While Xaibo's automatic dependency resolution reduces this burden, complex systems still require explicit configuration to specify component relationships. **Debugging Challenges**: When something goes wrong in a modular system, the problem might be in any of the components or in their interactions. Debugging requires understanding the flow of control across component boundaries, which can be more complex than debugging a monolithic system. ## Xaibo's Approach to Modularity Xaibo's approach to modularity is designed to maximize the benefits while minimizing the costs. Several design decisions contribute to this balance: **Protocol-Driven Interfaces**: By using protocols to define component interfaces, Xaibo ensures that modules can be composed flexibly while maintaining type safety and clear contracts. The protocol system provides the benefits of modularity without sacrificing the clarity of well-defined interfaces. **Automatic Dependency Resolution**: The exchange system automatically resolves dependencies in simple cases, reducing the configuration burden. You can often define modules and let the framework figure out how to connect them, getting the benefits of modularity without the complexity of explicit wiring. **Transparent Observability**: The proxy system provides comprehensive visibility into component interactions without requiring explicit instrumentation. This makes debugging modular systems easier by providing automatic insight into how components interact. **Comprehensive Testing Support**: The framework's testing support makes it easy to create mock implementations and test components in isolation. This addresses one of the traditional challenges of modular systems, ensuring that components work correctly both individually and in composition. ## Modularity Patterns in Xaibo Xaibo employs several specific modularity patterns that are worth understanding: **Layered Architecture**: Components are organized into layers based on their level of abstraction. Protocol definitions form the interface layer, primitive modules provide basic implementations, and composite modules provide higher-level behaviors. **Plugin Architecture**: New modules can be added to the system without modifying existing code. The protocol system provides stable interfaces that new modules can implement, enabling extensibility without coupling. **Dependency Injection**: Components declare their dependencies explicitly, allowing the framework to wire them together automatically. This pattern enables flexible composition while maintaining clear component boundaries. **Event-Driven Communication**: Components communicate through events as well as direct protocol calls. This enables loose coupling for cross-cutting concerns like logging, monitoring, and debugging. ## Comparison with Other Approaches Different AI frameworks have taken different approaches to the modularity question, each with their own trade-offs: **Monolithic Frameworks**: Provide comprehensive, integrated solutions that work well out of the box but are difficult to customize or extend. Examples include many commercial AI platforms that provide end-to-end solutions. **Library-Based Approaches**: Provide collections of utilities that can be composed into applications. These offer more flexibility than monolithic frameworks but require more work to integrate components and often lack consistent interfaces. **Microservice Architectures**: Decompose systems into independent services that communicate over networks. This provides strong isolation but adds significant operational complexity and performance overhead. **Plugin Architectures**: Provide core functionality with extension points for additional capabilities. This balances flexibility with simplicity but can be limiting when core assumptions need to change. Xaibo's approach combines aspects of these patterns while avoiding their main limitations. It provides the flexibility of library-based approaches with the consistency of monolithic frameworks, the isolation benefits of microservices without the operational overhead, and the extensibility of plugin architectures without the rigidity of fixed core assumptions. ## Modularity and AI System Evolution The rapid pace of evolution in AI technology makes modularity particularly valuable for AI systems. New model architectures, new training techniques, and new application patterns emerge regularly. A modular architecture enables systems to evolve with these changes rather than being locked into specific approaches. Consider how the AI landscape has changed in just the past few years: new model providers have emerged, function calling has become standard, vector databases have proliferated, and new interaction patterns like tool use have become common. A monolithic framework built two years ago would struggle to incorporate these changes, while a modular framework can adapt by adding new modules or updating existing ones. This evolutionary pressure is likely to continue, making modularity not just a nice-to-have feature but a necessity for long-term viability. Systems that can't adapt to changing technology will become obsolete, while modular systems can evolve continuously. ## The Network Effect of Modularity One of the most powerful aspects of modularity is the network effect it creates. As more modules are developed, the value of the entire ecosystem increases. Each new LLM module makes the framework more valuable to users who want to use that provider. Each new tool integration makes the framework more useful for applications that need those capabilities. This network effect is particularly important in the AI space, where the ecosystem is rapidly expanding. By providing a stable foundation for module development, Xaibo enables a community of developers to contribute modules that benefit everyone using the framework. The network effect also applies to knowledge and expertise. Developers who understand how to build modules for one protocol can easily build modules for other protocols. The skills and patterns learned in one context transfer to others, creating a multiplier effect for development effort. ## Modularity as a Design Philosophy Beyond its technical benefits, modularity represents a broader design philosophy that prioritizes flexibility, maintainability, and long-term thinking over short-term convenience. This philosophy recognizes that software systems evolve over time and that the decisions made early in a project's lifecycle will constrain options later. The modular approach reflects several important principles: **Separation of Concerns**: Each module has a focused responsibility, making it easier to understand, test, and maintain. **Open/Closed Principle**: The system is open for extension through new modules but closed for modification of existing components. **Dependency Inversion**: High-level modules don't depend on low-level modules; both depend on abstractions (protocols). **Single Responsibility**: Each module has one reason to change, reducing the risk that changes in one area will affect others. These principles aren't just academic concepts, they have practical implications for how systems behave over time. Modular systems tend to be more maintainable, more testable, and more adaptable to changing requirements. ## The Future of Modular AI Systems Xaibo's modular approach anticipates a future where AI systems are composed from interchangeable, specialized components rather than built as monolithic applications. This vision aligns with broader trends in software engineering toward microservices, containerization, and cloud-native architectures. As AI capabilities become more specialized and diverse, the ability to compose systems from focused components becomes increasingly valuable. A modular framework enables developers to build systems that are exactly suited to their needs rather than accepting the compromises inherent in one-size-fits-all solutions. The modular approach also enables new possibilities for AI system development, such as automatic optimization of component compositions, marketplace-style distribution of modules, and collaborative development of complex systems across organizational boundaries. By choosing modularity, Xaibo positions itself not just as a framework for building today's AI systems, but as a foundation for the AI systems of the future. The modular architecture provides the flexibility and extensibility needed to evolve with the rapidly changing AI landscape while maintaining the stability and reliability needed for production systems. # How Xaibo's Design Enables Superior Testing Testing AI systems presents unique challenges that traditional testing approaches struggle to address. The non-deterministic nature of language models, the complexity of multi-component interactions, and the reliance on external services create a perfect storm of testing difficulties. Xaibo's architecture is specifically designed to address these challenges, making comprehensive testing not just possible but natural and efficient. ## The AI Testing Problem Traditional software testing relies on predictability. You provide known inputs, expect specific outputs, and verify that the system behaves as expected. This approach works well for deterministic systems where the same input always produces the same output. AI systems violate this fundamental assumption. A language model might generate different responses to the same prompt, tool executions might fail due to network issues, and complex agent behaviors might emerge from the interaction of multiple non-deterministic components. These characteristics make traditional testing approaches inadequate. Consider testing an AI agent that uses a language model to analyze user input, retrieves relevant information from memory, executes tools based on the analysis, and generates a response. How do you write a test for this system? The language model's response is unpredictable, the memory retrieval depends on previous interactions, the tool execution might fail for external reasons, and the final response incorporates all these variables. Traditional approaches might try to mock the language model with fixed responses, but this doesn't test the real system behavior. They might try to test against live services, but this makes tests slow, expensive, and unreliable. They might try to test individual components in isolation, but this misses the emergent behaviors that arise from component interactions. ## Dependency Injection as a Testing Foundation Xaibo's dependency injection architecture provides the foundation for superior testing by making all dependencies explicit and replaceable. Instead of components creating their own dependencies or accessing them through global state, they receive dependencies through their constructors. This explicit dependency pattern enables a powerful testing strategy: replace production dependencies with test implementations that provide predictable, controllable behavior. Instead of testing against a real language model, inject a mock implementation that returns predetermined responses. Instead of using a real vector database, inject an in-memory implementation that provides consistent results. The key insight is that the component being tested doesn't know or care whether it's receiving production or test implementations. As long as the test implementations fulfill the same protocol contracts, the component behaves identically in both environments. ## Protocol-Based Mocking Xaibo's protocol-driven architecture makes mocking particularly elegant. Instead of creating complex mock objects that simulate the behavior of specific implementations, you can create simple test implementations that fulfill protocol contracts. Consider testing an orchestrator that depends on [`LLMProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/llm.py). Instead of mocking the OpenAI API or trying to simulate network responses, you create a simple test implementation: ```python class TestLLM: def __init__(self, responses: list[str]): self.responses = responses self.call_count = 0 async def generate(self, messages, options=None): response = self.responses[self.call_count % len(self.responses)] self.call_count += 1 return LLMResponse(content=response) ``` This test implementation provides complete control over the language model's behavior while implementing the same protocol that production LLM modules use. The orchestrator receives this test implementation through the same dependency injection mechanism used in production, ensuring that the test environment accurately reflects the production environment. ## Comprehensive Test Coverage The dependency injection approach enables comprehensive test coverage that would be difficult or impossible to achieve with traditional testing approaches. You can test: **Individual Component Logic**: Test each module in isolation by providing mock implementations of its dependencies. This enables focused testing of specific algorithms and behaviors without the complexity of external integrations. **Integration Patterns**: Test how components interact by providing real implementations of some dependencies and mock implementations of others. This enables testing of integration logic while maintaining control over external factors. **Error Handling**: Test error conditions by providing mock implementations that simulate various failure modes. This enables comprehensive testing of error handling logic without relying on external systems to fail in specific ways. **Performance Characteristics**: Test performance by providing mock implementations with controlled timing characteristics. This enables performance testing without the variability of external services. **Edge Cases**: Test edge cases by providing mock implementations that return unusual or boundary-condition responses. This enables testing of robustness without trying to trigger edge cases in external systems. ## The MockLLM Pattern Xaibo includes several built-in test implementations that demonstrate effective testing patterns. The [`MockLLM`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/primitives/modules/llm/mock.py) module provides a particularly instructive example of how to create effective test implementations. The `MockLLM` implements the full `LLMProtocol` while providing complete control over responses. It can return predetermined responses, simulate streaming behavior, introduce controlled delays, and even simulate error conditions. This enables testing of complex orchestration logic without the unpredictability of real language models. The pattern demonstrated by `MockLLM` can be applied to other protocols as well. Test implementations of [`ToolProviderProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/tools.py) can simulate tool execution without actually running external commands. Test implementations of [`MemoryProtocol`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/protocols/memory.py) can provide predictable memory behavior without external databases. ## Event-Driven Test Verification Xaibo's event system provides another powerful testing capability: the ability to verify not just what happened, but how it happened. Events capture the complete flow of execution, including which components were called, what parameters were passed, and how long operations took. This event-driven approach enables sophisticated test verification strategies: **Behavioral Verification**: Verify that components interact in expected ways by examining the sequence of events generated during test execution. **Performance Verification**: Verify that operations complete within expected time bounds by examining timing information in events. **Integration Verification**: Verify that complex workflows execute correctly by examining the complete event trace. **Regression Detection**: Compare event traces from different test runs to detect changes in system behavior. The event system also enables automatic test case generation. By capturing events during manual testing or production operation, you can generate test cases that verify the same behavior in automated tests. ## Configuration Override Testing Xaibo's configuration override system provides fine-grained control over test environments. The [`ConfigOverrides`](https://github.com/xpressai/xaibo/blob/main/src/xaibo/core/config.py) mechanism allows you to replace specific components while leaving others unchanged. This capability enables sophisticated testing strategies: **Partial Mocking**: Replace only the components that need to be controlled for testing while using real implementations for others. **A/B Testing**: Compare different implementations of the same protocol by running the same test with different component configurations. **Environment Simulation**: Simulate different deployment environments by providing different implementations of infrastructure-related protocols. **Failure Injection**: Inject failing implementations of specific protocols to test error handling and recovery logic. ## Test Performance and Reliability One of the most significant benefits of Xaibo's testing approach is the dramatic improvement in test performance and reliability. By replacing external dependencies with local test implementations, tests become: **Fast**: No network calls, no external service dependencies, no waiting for language model inference. **Reliable**: No flaky network connections, no service outages, no rate limiting. **Deterministic**: Predictable responses, consistent timing, reproducible behavior. **Isolated**: Tests don't interfere with each other or depend on external state. These characteristics enable comprehensive test suites that can run quickly and reliably in continuous integration environments. Developers can run extensive tests locally without worrying about external dependencies or service costs. ## Testing Complex Behaviors Xaibo's architecture enables testing of complex, emergent behaviors that would be difficult to test in traditional architectures. By controlling the behavior of individual components through dependency injection, you can create scenarios that test sophisticated agent behaviors. For example, you might test an agent's ability to handle tool execution failures by providing a tool implementation that fails in specific ways. You might test memory retrieval logic by providing a memory implementation with specific content. You might test conversation flow by providing an LLM implementation that responds in particular patterns. These controlled scenarios enable testing of edge cases and error conditions that might be rare or difficult to reproduce in production environments. The ability to create these scenarios reliably makes it possible to build robust, well-tested agents. ## Comparison with Traditional AI Testing Traditional approaches to testing AI systems often fall into several problematic patterns: **End-to-End Testing Only**: Testing the entire system as a black box, which makes it difficult to isolate issues and understand what's being tested. **Manual Testing**: Relying on human evaluation of AI outputs, which is time-consuming, subjective, and doesn't scale. **Production Testing**: Testing against live services, which is expensive, slow, and unreliable. **Snapshot Testing**: Capturing outputs and comparing them to previous runs, which breaks when AI behavior changes (which it often does). **Statistical Testing**: Running many tests and looking for statistical patterns, which is complex and doesn't provide clear pass/fail criteria. Xaibo's approach avoids these problems by enabling focused, deterministic testing of specific behaviors while maintaining the ability to test complex interactions and emergent behaviors. ## Test-Driven Development for AI Xaibo's testing capabilities enable test-driven development (TDD) practices for AI systems. You can write tests that specify desired behavior before implementing the components that provide that behavior. This approach provides several benefits: **Clear Requirements**: Tests serve as executable specifications of what the system should do. **Design Feedback**: Writing tests first provides feedback about component interfaces and dependencies. **Regression Protection**: Tests prevent changes from breaking existing functionality. **Refactoring Confidence**: Comprehensive tests enable confident refactoring and optimization. TDD is particularly valuable for AI systems because it forces you to think clearly about what behavior you want before getting lost in the complexity of implementation details. ## Testing as Documentation Well-written tests serve as documentation of how the system is intended to work. Xaibo's testing approach makes this documentation particularly valuable because tests can demonstrate not just individual component behavior, but complex interaction patterns. Test cases that show how an orchestrator coordinates multiple components, how error conditions are handled, or how different configurations affect behavior provide valuable documentation for developers trying to understand or extend the system. The event traces captured during testing also serve as documentation of system behavior, showing exactly how components interact to produce specific outcomes. ## Continuous Testing and Monitoring Xaibo's testing architecture enables continuous testing strategies that blur the line between testing and monitoring. The same test implementations used in development can be deployed in production environments to provide ongoing verification of system behavior. This approach enables: **Canary Testing**: Deploy new components alongside test implementations to verify behavior before switching to production traffic. **Regression Monitoring**: Continuously run tests against production systems to detect when behavior changes unexpectedly. **Performance Monitoring**: Use test implementations with known performance characteristics to detect performance regressions. **Health Checking**: Use simple test scenarios to verify that all system components are functioning correctly. ## The Future of AI Testing Xaibo's testing approach anticipates a future where AI systems are tested as rigorously as traditional software systems. As AI becomes more critical to business operations, the need for reliable, comprehensive testing becomes paramount. The framework's testing capabilities also enable new possibilities for AI system development, such as: **Automated Test Generation**: Using AI to generate test cases based on system behavior. **Behavioral Regression Detection**: Using machine learning to detect when system behavior changes in unexpected ways. **Performance Optimization**: Using test data to automatically optimize system configurations. **Quality Assurance**: Using comprehensive testing to provide confidence in AI system reliability. By making testing a first-class concern in AI system architecture, Xaibo enables the development of AI systems that are not just functional, but reliable, maintainable, and trustworthy. The testing capabilities aren't an afterthought, they're an integral part of the framework's design philosophy that prioritizes quality and reliability alongside functionality. ## Putting It Into Practice To see these testing concepts in action, work through the [Testing Agents tutorial](../../../tutorial/testing-agents/), which demonstrates hands-on implementation of dependency injection, event capture, and comprehensive agent testing strategies.