Create main.py

main.py

@@ -0,0 +1,161 @@
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, Field
+from typing import List, Optional, Literal, Dict, Any, Union
+import uvicorn
+from duckai import DuckAI
+
+app = FastAPI(title="DuckAI OpenAI Compatible API")
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Models for the OpenAI-compatible API
+class Message(BaseModel):
+    role: Literal["system", "user", "assistant"]
+    content: str
+
+class ChatCompletionRequest(BaseModel):
+    model: str
+    messages: List[Message]
+    temperature: Optional[float] = 1.0
+    max_tokens: Optional[int] = None
+    stream: Optional[bool] = False
+
+class ChatCompletionChoice(BaseModel):
+    index: int = 0
+    message: Message
+    finish_reason: str = "stop"
+
+class ChatCompletionUsage(BaseModel):
+    prompt_tokens: int
+    completion_tokens: int
+    total_tokens: int
+
+class ChatCompletionResponse(BaseModel):
+    id: str
+    object: str = "chat.completion"
+    created: int
+    model: str
+    choices: List[ChatCompletionChoice]
+    usage: ChatCompletionUsage
+
+# DuckAI Parser
+class DuckAIParser:
+    @staticmethod
+    def parse_conversation_history(messages: List[Message]) -> str:
+        """
+        Convert OpenAI message format to DuckAI's expected format with user: and assistant: prefixes
+        """
+        # Skip system messages as they aren't part of the core conversation
+        conversation = []
+        
+        for msg in messages:
+            if msg.role != "system":
+                conversation.append(f"{msg.role}: {msg.content}")
+        
+        return "\n".join(conversation)
+    
+    @staticmethod
+    def estimate_tokens(text: str) -> int:
+        """
+        Estimate token count in text - rough approximation
+        """
+        # Very simple estimation - about 4 characters per token on average
+        return len(text) // 4
+
+@app.post("/v1/chat/completions", response_model=ChatCompletionResponse)
+async def create_chat_completion(request: ChatCompletionRequest):
+    try:
+        # Parse the conversation history into DuckAI format
+        conversation_text = DuckAIParser.parse_conversation_history(request.messages)
+        
+        # Call DuckAI with the formatted conversation
+        duck_ai = DuckAI()
+        result = duck_ai.chat(conversation_text, model=request.model)
+        
+        # Extract the assistant's response
+        assistant_response = result.strip()
+        
+        # Estimate token usage
+        prompt_tokens = DuckAIParser.estimate_tokens(conversation_text)
+        completion_tokens = DuckAIParser.estimate_tokens(assistant_response)
+        
+        # Create OpenAI-compatible response
+        response = ChatCompletionResponse(
+            id=f"chatcmpl-duck-{hash(conversation_text) % 10000}",
+            created=int(__import__('time').time()),
+            model=request.model,
+            choices=[
+                ChatCompletionChoice(
+                    message=Message(
+                        role="assistant",
+                        content=assistant_response
+                    )
+                )
+            ],
+            usage=ChatCompletionUsage(
+                prompt_tokens=prompt_tokens,
+                completion_tokens=completion_tokens,
+                total_tokens=prompt_tokens + completion_tokens
+            )
+        )
+        
+        return response
+    
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/v1/models")
+async def list_models():
+    """Return a list of available models"""
+    current_time = int(__import__('time').time())
+    return {
+        "object": "list",
+        "data": [
+            {
+                "id": "gpt-4o-mini",
+                "object": "model",
+                "created": current_time,
+                "owned_by": "DuckAI"
+            },
+            {
+                "id": "llama-3.3-70b",
+                "object": "model",
+                "created": current_time,
+                "owned_by": "DuckAI"
+            },
+            {
+                "id": "claude-3-haiku",
+                "object": "model",
+                "created": current_time,
+                "owned_by": "DuckAI"
+            },
+            {
+                "id": "o3-mini",
+                "object": "model",
+                "created": current_time,
+                "owned_by": "DuckAI"
+            },
+            {
+                "id": "mistral-small-3",
+                "object": "model",
+                "created": current_time,
+                "owned_by": "DuckAI"
+            }
+        ]
+    }
+
+# Adding a simple root endpoint for health check
+@app.get("/")
+async def root():
+    return {"status": "ok", "message": "DuckAI OpenAI Compatible API is running"}
+
+if __name__ == "__main__":
+    uvicorn.run("main:app", host="0.0.0.0", port=7860, reload=True)