import os import logging import json import time # Import time module from aiohttp import web from aiohttp_cors import setup as cors_setup, ResourceOptions from dotenv import load_dotenv # Load environment variables from .env file load_dotenv() # Set up logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s') logger = logging.getLogger(__name__) # Import the agent logic from src.agents import ( get_simple_litellm_agent, get_ollama_native_agent, OLLAMA_NATIVE_AVAILABLE, get_pydantic_agent, PYDANTIC_AI_AVAILABLE ) from src.tools import get_all_tools from src.utils import extract_user_id_from_request async def chat_completions(request): """ Handles OpenAI-compatible chat completion requests using Ollama Native agent. Default endpoint - uses Ollama Native Agent with tools enabled. """ if not OLLAMA_NATIVE_AVAILABLE: return web.json_response({ "error": {"message": "Ollama Native agent not available"} }, status=503) try: data = await request.json() logger.info(f"[DEFAULT/OLLAMA_NATIVE] Received chat request") # Extract relevant fields from the request messages = data.get("messages") model = data.get("model", "ollama-native") stream = data.get("stream", False) conversation_id = data.get("conversation_id") enable_tools = data.get("enable_tools", True) # Tools enabled by default if not messages: raise web.HTTPBadRequest(reason="'messages' field is required") # Get the agent instance (Ollama Native with working tool calling) agent = get_ollama_native_agent(discover_tools=enable_tools) # For non-streaming requests, collect the full response if not stream: response_content = await agent.chat_completion( messages=messages, conversation_id=conversation_id ) return web.json_response({ "id": f"chatcmpl-{os.urandom(12).hex()}", "object": "chat.completion", "created": int(time.time()), "model": "pydantic", "choices": [{ "index": 0, "message": {"role": "assistant", "content": response_content}, "finish_reason": "stop" }], "usage": { "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0 }, "tools_enabled": enable_tools, "tools_count": len(agent.tools_dict) if enable_tools else 0 }) else: # Handle streaming response response = web.StreamResponse( status=200, headers={'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive'} ) await response.prepare(request) async for chunk in agent.chat(messages=messages, conversation_id=conversation_id, stream=True): chunk_type = chunk.get("type", "content") if chunk_type == "content": json_chunk = { "id": f"chatcmpl-{os.urandom(12).hex()}", "object": "chat.completion.chunk", "created": int(time.time()), "model": "pydantic", "choices": [{ "index": 0, "delta": {"content": chunk.get("content", "")}, "finish_reason": chunk.get("finish_reason") }] } await response.write(f"data: {json.dumps(json_chunk)}\n\n".encode()) if chunk.get("finish_reason") == "stop": break elif chunk_type == "error": error_chunk = { "error": {"message": chunk.get("content", "Unknown error")} } await response.write(f"data: {json.dumps(error_chunk)}\n\n".encode()) break await response.write(b"data: [DONE]\n\n") await response.write_eof() return response except web.HTTPBadRequest as e: logger.warning(f"Bad request: {e.reason}") return web.json_response({"error": {"message": e.reason}}, status=400) except Exception as e: logger.exception("[DEFAULT/PYDANTIC_AI] Error during chat completion:") return web.json_response({"error": {"message": str(e)}}, status=500) async def chat_simple(request): """ Handles chat requests using SimpleLiteLLMAgent (no tools). Endpoint: POST /v1/chat/simple """ try: data = await request.json() logger.info(f"[SIMPLE] Received chat request") messages = data.get("messages") model = data.get("model", "simple") stream = data.get("stream", False) conversation_id = data.get("conversation_id") if not messages: raise web.HTTPBadRequest(reason="'messages' field is required") # Get SimpleLiteLLM agent agent = get_simple_litellm_agent() # Non-streaming response if not stream: response_content = await agent.chat_completion( messages=messages, conversation_id=conversation_id ) return web.json_response({ "id": f"chatcmpl-{os.urandom(12).hex()}", "object": "chat.completion", "created": int(time.time()), "model": "simple", "choices": [{ "index": 0, "message": {"role": "assistant", "content": response_content}, "finish_reason": "stop" }], "usage": { "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0 } }) else: # Streaming response response = web.StreamResponse( status=200, headers={'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive'} ) await response.prepare(request) async for chunk in agent.chat(messages=messages, conversation_id=conversation_id, stream=True): json_chunk = { "id": f"chatcmpl-{os.urandom(12).hex()}", "object": "chat.completion.chunk", "created": int(time.time()), "model": "simple", "choices": [{ "index": 0, "delta": {"content": chunk.get("content", "")}, "finish_reason": chunk.get("finish_reason") }] } await response.write(f"data: {json.dumps(json_chunk)}\n\n".encode()) if chunk.get("finish_reason") == "stop": break await response.write(b"data: [DONE]\n\n") await response.write_eof() return response except web.HTTPBadRequest as e: logger.warning(f"Bad request: {e.reason}") return web.json_response({"error": {"message": e.reason}}, status=400) except Exception as e: logger.exception("[SIMPLE] Error during chat completion:") return web.json_response({"error": {"message": str(e)}}, status=500) async def chat_pydantic(request): """ Handles chat requests using PydanticAI Agent with tools. Endpoint: POST /v1/chat/pydantic """ if not PYDANTIC_AI_AVAILABLE: return web.json_response({ "error": {"message": "PydanticAI not available. Install with: pip install pydantic-ai"} }, status=503) try: data = await request.json() logger.info(f"[PYDANTIC_AI] Received chat request") messages = data.get("messages") model = data.get("model", "pydantic") stream = data.get("stream", False) conversation_id = data.get("conversation_id") enable_tools = data.get("enable_tools", True) # Extract user ID from request user_id = extract_user_id_from_request(data) if not messages: raise web.HTTPBadRequest(reason="'messages' field is required") # Get PydanticAI agent with or without tools agent = get_pydantic_agent(discover_tools=enable_tools, user_id=user_id) # Non-streaming response if not stream: response_content = await agent.chat_completion( messages=messages, conversation_id=conversation_id ) return web.json_response({ "id": f"chatcmpl-{os.urandom(12).hex()}", "object": "chat.completion", "created": int(time.time()), "model": "pydantic", "choices": [{ "index": 0, "message": {"role": "assistant", "content": response_content}, "finish_reason": "stop" }], "usage": { "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0 }, "tools_enabled": enable_tools, "tools_count": len(agent.tools_dict) if enable_tools else 0 }) else: # Streaming response response = web.StreamResponse( status=200, headers={'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive'} ) await response.prepare(request) async for chunk in agent.chat(messages=messages, conversation_id=conversation_id, stream=True): chunk_type = chunk.get("type", "content") if chunk_type == "content": json_chunk = { "id": f"chatcmpl-{os.urandom(12).hex()}", "object": "chat.completion.chunk", "created": int(time.time()), "model": "pydantic", "choices": [{ "index": 0, "delta": {"content": chunk.get("content", "")}, "finish_reason": chunk.get("finish_reason") }] } await response.write(f"data: {json.dumps(json_chunk)}\n\n".encode()) if chunk.get("finish_reason") == "stop": break elif chunk_type == "error": error_chunk = { "error": {"message": chunk.get("content", "Unknown error")} } await response.write(f"data: {json.dumps(error_chunk)}\n\n".encode()) break await response.write(b"data: [DONE]\n\n") await response.write_eof() return response except web.HTTPBadRequest as e: logger.warning(f"Bad request: {e.reason}") return web.json_response({"error": {"message": e.reason}}, status=400) except Exception as e: logger.exception("[PYDANTIC_AI] Error during chat completion:") return web.json_response({"error": {"message": str(e)}}, status=500) async def list_models(request): """ Lists available models (OpenAI-compatible endpoint). Endpoint: GET /v1/models """ models = [ { "id": "Tatlock", "object": "model", "created": int(time.time()), "owned_by": "core-ai", "permission": [], "root": "tatlock", "parent": None, }, { "id": "simple", "object": "model", "created": int(time.time()), "owned_by": "core-ai", "permission": [], "root": "simple", "parent": None, } ] return web.json_response({ "object": "list", "data": models }) async def list_tools(request): """ Lists all available tools. Endpoint: GET /v1/tools """ try: tools = get_all_tools() tools_info = [] for name, func in tools.items(): tools_info.append({ "name": name, "description": func.__doc__.strip() if func.__doc__ else "No description available", "type": "local" }) return web.json_response({ "tools": tools_info, "count": len(tools_info), "pydantic_ai_available": PYDANTIC_AI_AVAILABLE }) except Exception as e: logger.exception("Error listing tools:") return web.json_response({"error": {"message": str(e)}}, status=500) async def health_check(request): """Simple health check endpoint.""" return web.json_response({ "status": "ok", "service": "core-ai", "agents": { "simple": True, "ollama-native": OLLAMA_NATIVE_AVAILABLE, "pydantic": PYDANTIC_AI_AVAILABLE }, "default_agent": "ollama-native" if OLLAMA_NATIVE_AVAILABLE else "simple", "tools_count": len(get_all_tools()) }) async def test_ollama_tools(request): """Test Ollama tool calling directly""" import httpx try: tool_def = { "type": "function", "function": { "name": "web_search", "description": "Search the web using SearXNG", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Search query"} }, "required": ["query"] } } } payload = { "model": "mistral-nemo:latest", "messages": [{"role": "user", "content": "Search for Python 3.13 features"}], "tools": [tool_def], "stream": False } async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post('http://ollama:11434/api/chat', json=payload) result = response.json() return web.json_response({ "status_code": response.status_code, "has_tool_calls": 'tool_calls' in result.get('message', {}), "response": result }) except Exception as e: logger.exception("Test error:") return web.json_response({"error": str(e)}, status=500) async def setup_routes(app): # Chat endpoints app.router.add_post("/chat/completions", chat_completions) # Alias without /v1 for compatibility app.router.add_post("/v1/chat/completions", chat_completions) # Default (PydanticAI) app.router.add_post("/v1/chat/simple", chat_simple) # Simple agent (no tools) app.router.add_post("/v1/chat/pydantic", chat_pydantic) # Alias for default # OpenAI-compatible endpoints app.router.add_get("/v1/models", list_models) # List available models app.router.add_get("/models", list_models) # Alias without /v1 prefix # Tool management app.router.add_get("/v1/tools", list_tools) # List available tools # Health check app.router.add_get("/health", health_check) app.router.add_get("/test/ollama-tools", test_ollama_tools) # Setup CORS cors = cors_setup(app, defaults={ "*": ResourceOptions( allow_credentials=True, expose_headers="*", allow_headers="*", allow_methods="*" ) }) # Configure CORS on all routes for route in list(app.router.routes()): cors.add(route) def main(): app = web.Application() app.on_startup.append(setup_routes) # Register routes on startup # Configuration host = os.getenv("HOST", "0.0.0.0") port = int(os.getenv("PORT", 8086)) # Use 8086 to avoid conflict with core-ai logger.info(f"Starting core-ai service on http://{host}:{port}") web.run_app(app, host=host, port=port) if __name__ == "__main__": main()