Upload app.py with huggingface_hub

app.py

@@ -1,25 +1,443 @@
-from fastapi import FastAPI
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
 from datetime import datetime
+from typing import List, Dict, Any, Optional
+import json
+import re
 
-app = FastAPI()
+app = FastAPI(title="AI Agent with Anthropic/OpenAI SDK Support", version="2.0.0")
 
+# Enable CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Data Models for SDK Compatibility
+class ToolDefinition(BaseModel):
+    name: str
+    description: str
+    parameters: Dict[str, Any]
+
+class AnthropicRequest(BaseModel):
+    messages: List[Dict[str, Any]]
+    tools: Optional[List[ToolDefinition]] = None
+    model: Optional[str] = "claude-3-sonnet-20240229"
+    max_tokens: Optional[int] = 1024
+    temperature: Optional[float] = 0.7
+    reasoning_split: Optional[bool] = False
+
+class OpenAIRequest(BaseModel):
+    messages: List[Dict[str, Any]]
+    tools: Optional[List[ToolDefinition]] = None
+    model: Optional[str] = "gpt-4"
+    max_tokens: Optional[int] = 1024
+    temperature: Optional[float] = 0.7
+    reasoning_split: Optional[bool] = False
+
+class AgenticRequest(BaseModel):
+    task: str
+    context: Optional[str] = ""
+    conversation_history: Optional[List[Dict[str, Any]]] = []
+    tools: Optional[List[ToolDefinition]] = None
+    reasoning_split: Optional[bool] = False
+
+# Available Tools
+AVAILABLE_TOOLS = {
+    "search_web": {
+        "name": "search_web",
+        "description": "Search the web for information based on a query",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "query": {"type": "string", "description": "Search query to look up information"}
+            },
+            "required": ["query"]
+        }
+    },
+    "analyze_code": {
+        "name": "analyze_code",
+        "description": "Analyze code for understanding, structure, and suggestions",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "code": {"type": "string", "description": "Code to analyze"},
+                "task": {"type": "string", "description": "Analysis task (comprehensive_analysis, optimization, debugging)"}
+            },
+            "required": ["code"]
+        }
+    },
+    "get_weather": {
+        "name": "get_weather",
+        "description": "Get current weather information for a location",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "location": {"type": "string", "description": "Location to get weather for"},
+                "units": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"}
+            },
+            "required": ["location"]
+        }
+    },
+    "math_calc": {
+        "name": "math_calc",
+        "description": "Perform mathematical calculations",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "expression": {"type": "string", "description": "Mathematical expression to calculate"},
+                "precision": {"type": "integer", "default": 6, "description": "Number of decimal places"}
+            },
+            "required": ["expression"]
+        }
+    },
+    "text_analyzer": {
+        "name": "text_analyzer",
+        "description": "Analyze text for sentiment, keywords, and insights",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "text": {"type": "string", "description": "Text to analyze"},
+                "analysis_type": {"type": "string", "enum": ["sentiment", "keywords", "summary"], "default": "sentiment"}
+            },
+            "required": ["text"]
+        }
+    }
+}
+
+def determine_tools_needed(task: str) -> List[Dict[str, Any]]:
+    """Determine which tools are needed based on task analysis"""
+    task_lower = task.lower()
+    selected_tools = []
+    
+    # Search tools
+    if any(keyword in task_lower for keyword in ["search", "find", "look up", "research", "information"]):
+        selected_tools.append(AVAILABLE_TOOLS["search_web"])
+    
+    # Code analysis tools
+    if any(keyword in task_lower for keyword in ["code", "function", "program", "debug", "optimize"]):
+        selected_tools.append(AVAILABLE_TOOLS["analyze_code"])
+    
+    # Weather tools
+    if any(keyword in task_lower for keyword in ["weather", "temperature", "climate"]):
+        selected_tools.append(AVAILABLE_TOOLS["get_weather"])
+    
+    # Math tools
+    if any(keyword in task_lower for keyword in ["calculate", "math", "+", "-", "*", "/", "="]) or re.search(r'\d+\s*[+\-*/=]\s*\d+', task_lower):
+        selected_tools.append(AVAILABLE_TOOLS["math_calc"])
+    
+    # Text analysis tools
+    if any(keyword in task_lower for keyword in ["analyze", "sentiment", "summary", "text"]):
+        selected_tools.append(AVAILABLE_TOOLS["text_analyzer"])
+    
+    return selected_tools
+
+def generate_tool_calls(task: str, context: str = "") -> List[Dict[str, Any]]:
+    """Generate appropriate tool calls based on task"""
+    tool_calls = []
+    current_time = datetime.now().strftime('%Y%m%d_%H%M%S')
+    
+    if "search" in task.lower() or "find" in task.lower():
+        query = task.replace("search for", "").replace("find", "").replace("look up", "").strip()
+        tool_calls.append({
+            "id": f"search_{current_time}",
+            "function": {
+                "name": "search_web",
+                "arguments": json.dumps({"query": query or task})
+            }
+        })
+    
+    elif "code" in task.lower():
+        tool_calls.append({
+            "id": f"code_analysis_{current_time}",
+            "function": {
+                "name": "analyze_code",
+                "arguments": json.dumps({
+                    "code": context or "Sample code for analysis",
+                    "task": "comprehensive_analysis"
+                })
+            }
+        })
+    
+    elif "weather" in task.lower():
+        location = context or "current location"
+        tool_calls.append({
+            "id": f"weather_{current_time}",
+            "function": {
+                "name": "get_weather",
+                "arguments": json.dumps({"location": location})
+            }
+        })
+    
+    elif any(char.isdigit() for char in task) and any(op in task for op in ["+", "-", "*", "/"]):
+        tool_calls.append({
+            "id": f"math_{current_time}",
+            "function": {
+                "name": "math_calc",
+                "arguments": json.dumps({"expression": task})
+            }
+        })
+    
+    elif "analyze" in task.lower():
+        tool_calls.append({
+            "id": f"text_analysis_{current_time}",
+            "function": {
+                "name": "text_analyzer",
+                "arguments": json.dumps({
+                    "text": context or task,
+                    "analysis_type": "sentiment"
+                })
+            }
+        })
+    
+    return tool_calls
+
+def create_anthropic_response(task: str, reasoning_split: bool = False) -> Dict[str, Any]:
+    """Create Anthropic SDK compatible response"""
+    tool_calls = generate_tool_calls(task)
+    reasoning = f"Analyzing task: '{task}'. I need to determine the best approach. "
+    
+    if tool_calls:
+        reasoning += f"I will use the {tool_calls[0]['function']['name']} tool to help with this task."
+    else:
+        reasoning += "This appears to be a general reasoning task that doesn't require specific tools."
+    
+    content_blocks = []
+    
+    if reasoning_split:
+        content_blocks.append({
+            "type": "thinking",
+            "text": reasoning
+        })
+    
+    if tool_calls:
+        content_blocks.append({
+            "type": "tool_use",
+            "id": tool_calls[0]["id"],
+            "name": tool_calls[0]["function"]["name"],
+            "input": json.loads(tool_calls[0]["function"]["arguments"])
+        })
+    
+    content_blocks.append({
+        "type": "text",
+        "text": f"I've analyzed your task: '{task}'. I'm ready to proceed with the best approach."
+    })
+    
+    return {
+        "id": f"msg_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
+        "type": "message",
+        "role": "assistant",
+        "content": content_blocks,
+        "model": "claude-3-sonnet-20240229",
+        "stop_reason": "tool_use" if tool_calls else "end_turn"
+    }
+
+def create_openai_response(task: str, reasoning_split: bool = False) -> Dict[str, Any]:
+    """Create OpenAI SDK compatible response"""
+    tool_calls = generate_tool_calls(task)
+    
+    reasoning = f"I'm analyzing the task: '{task}'. "
+    if tool_calls:
+        reasoning += f"I need to use the {tool_calls[0]['function']['name']} tool to help complete this."
+    else:
+        reasoning += "This is a reasoning task that requires my cognitive abilities."
+    
+    return {
+        "id": f"chatcmpl-{datetime.now().strftime('%Y%m%d_%H%M%S')}",
+        "object": "chat.completion",
+        "created": int(datetime.now().timestamp()),
+        "model": "gpt-4",
+        "choices": [
+            {
+                "index": 0,
+                "message": {
+                    "role": "assistant",
+                    "content": f"Working on your task: '{task}'. Let me proceed with the analysis." if not reasoning_split else None,
+                    "function_call": None,
+                    "tool_calls": tool_calls if tool_calls else None
+                },
+                "finish_reason": "tool_calls" if tool_calls else "stop"
+            }
+        ],
+        "usage": {
+            "prompt_tokens": 10,
+            "completion_tokens": 50,
+            "total_tokens": 60
+        },
+        "reasoning_details": reasoning if reasoning_split else None
+    }
+
+# API Endpoints
 @app.get("/")
 async def root():
-    return {"message": "AI Agent Space", "status": "running", "timestamp": datetime.now().isoformat()}
+    return {
+        "message": "AI Agent with Anthropic/OpenAI SDK Support",
+        "version": "2.0.0",
+        "status": "running",
+        "timestamp": datetime.now().isoformat(),
+        "endpoints": [
+            "/",
+            "/health", 
+            "/agentic",
+            "/anthropic/compatible",
+            "/openai/compatible"
+        ]
+    }
 
 @app.get("/health")
 async def health():
-    return {"status": "healthy"}
+    return {"status": "healthy", "timestamp": datetime.now().isoformat()}
+
+@app.get("/models")
+async def get_models():
+    return {
+        "models": [
+            {
+                "name": "anthropic-claude",
+                "status": "loaded",
+                "description": "Anthropic SDK compatible agentic model"
+            },
+            {
+                "name": "openai-gpt4", 
+                "status": "loaded",
+                "description": "OpenAI SDK compatible agentic model"
+            }
+        ]
+    }
+
+@app.post("/agentic")
+async def agentic_endpoint(request: AgenticRequest):
+    """General agentic endpoint with interleaved thinking"""
+    tools = request.tools or list(AVAILABLE_TOOLS.values())
+    tool_calls = generate_tool_calls(request.task, request.context)
+    
+    reasoning = f"I need to analyze the task: '{request.task}'. "
+    if tool_calls:
+        reasoning += f"This task requires using the {tool_calls[0]['function']['name']} tool. "
+    
+    reasoning += "I'm reasoning through each step to determine the best approach for tool use."
+    
+    return {
+        "reasoning_details": reasoning if request.reasoning_split else None,
+        "content": f"Working on your task: '{request.task}'. I have the tools I need to assist.",
+        "tool_calls": tool_calls,
+        "status": "success",
+        "thinking_process": "Interleaved thinking enabled - reasoning performed between tool interactions",
+        "tools_available": len(tools),
+        "conversation_length": len(request.conversation_history) if request.conversation_history else 0
+    }
 
-@app.get("/agentic")
-async def agentic():
+@app.post("/anthropic/compatible")
+async def anthropic_endpoint(request: AnthropicRequest):
+    """Anthropic SDK compatible endpoint with interleaved thinking"""
+    # Get the latest user message
+    latest_message = request.messages[-1] if request.messages else {"content": "Hello"}
+    user_content = latest_message.get("content", "")
+    
+    if isinstance(user_content, list):
+        user_text = " ".join([item.get("text", "") for item in user_content if item.get("type") == "text"])
+    else:
+        user_text = str(user_content)
+    
+    response = create_anthropic_response(user_text, request.reasoning_split)
+    
+    # Add available tools if specified
+    if request.tools:
+        available_tools = [tool for tool in AVAILABLE_TOOLS.values() if tool["name"] in [t.name for t in request.tools]]
+    else:
+        available_tools = list(AVAILABLE_TOOLS.values())
+    
     return {
-        "reasoning_details": "Interleaved thinking enabled",
-        "content": "Agentic model ready",
-        "tool_calls": [],
-        "status": "success"
+        **response,
+        "available_tools": available_tools,
+        "reasoning_split": request.reasoning_split
     }
 
+@app.post("/openai/compatible")
+async def openai_endpoint(request: OpenAIRequest):
+    """OpenAI SDK compatible endpoint with interleaved thinking"""
+    # Get the latest user message
+    latest_message = request.messages[-1] if request.messages else {"content": "Hello"}
+    user_content = latest_message.get("content", "")
+    
+    if isinstance(user_content, list):
+        user_text = " ".join([item.get("text", "") for item in user_content if item.get("type") == "text"])
+    else:
+        user_text = str(user_content)
+    
+    response = create_openai_response(user_text, request.reasoning_split)
+    
+    # Add available tools if specified
+    if request.tools:
+        available_tools = [tool for tool in AVAILABLE_TOOLS.values() if tool["name"] in [t.name for t in request.tools]]
+    else:
+        available_tools = list(AVAILABLE_TOOLS.values())
+    
+    return {
+        **response,
+        "available_tools": available_tools,
+        "reasoning_split": request.reasoning_split
+    }
+
+@app.get("/tools")
+async def list_tools():
+    """List all available tools"""
+    return {
+        "tools": list(AVAILABLE_TOOLS.values()),
+        "count": len(AVAILABLE_TOOLS)
+    }
+
+@app.post("/tools/search")
+async def simulate_tool_use(tool_name: str, arguments: Dict[str, Any]):
+    """Simulate tool use for demonstration"""
+    if tool_name not in AVAILABLE_TOOLS:
+        raise HTTPException(status_code=400, detail=f"Tool '{tool_name}' not found")
+    
+    # Simulate tool responses
+    if tool_name == "search_web":
+        return {
+            "tool": tool_name,
+            "result": f"Search results for '{arguments.get('query', 'unknown')}': Found relevant information.",
+            "status": "success"
+        }
+    elif tool_name == "analyze_code":
+        return {
+            "tool": tool_name,
+            "result": "Code analysis complete: Found 2 functions, 1 class, good structure overall.",
+            "status": "success"
+        }
+    elif tool_name == "get_weather":
+        return {
+            "tool": tool_name,
+            "result": f"Weather for {arguments.get('location', 'unknown')}: 22°C, partly cloudy",
+            "status": "success"
+        }
+    elif tool_name == "math_calc":
+        try:
+            expression = arguments.get("expression", "0")
+            result = eval(expression)
+            return {
+                "tool": tool_name,
+                "result": f"Calculation result: {result}",
+                "status": "success"
+            }
+        except:
+            return {
+                "tool": tool_name,
+                "result": "Error: Invalid mathematical expression",
+                "status": "error"
+            }
+    elif tool_name == "text_analyzer":
+        return {
+            "tool": tool_name,
+            "result": "Text analysis complete: Positive sentiment, found 3 key topics",
+            "status": "success"
+        }
+
 if __name__ == "__main__":
     import uvicorn
     uvicorn.run(app, host="0.0.0.0", port=7860)