route.py

gistfile1.txt

"""
FastAPI Intent-Based Tool Execution API

This API receives a user query, matches it to the most relevant intent,
and executes the appropriate tool using LangChain.
"""

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional, Dict, Any, List
import uvicorn
from datetime import datetime

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# Initialize FastAPI app
app = FastAPI(
    title="Intent-Based Tool Execution API",
    description="API that matches queries to intents and executes appropriate tools",
    version="1.0.0"
)

# Request/Response Models
class QueryRequest(BaseModel):
    """Request model for query endpoint"""
    query: str = Field(..., description="User's natural language query")
    user_id: Optional[str] = Field(None, description="Optional user identifier")

class IntentMatch(BaseModel):
    """Intent matching result"""
    intent: str = Field(..., description="Matched intent name")
    confidence: float = Field(..., description="Confidence score (0-1)")
    parameters: Dict[str, Any] = Field(default_factory=dict, description="Extracted parameters")

class QueryResponse(BaseModel):
    """Response model for query endpoint"""
    query: str
    intent_match: IntentMatch
    result: Any
    execution_time: float
    timestamp: str

# Define Tools using LangChain
@tool
def get_weather(location: str) -> str:
    """Get the current weather for a given location.

    Args:
        location: The city or location to get weather for
    """
    # Simulate weather API call
    return f"The weather in {location} is sunny and 72°F (22°C) with light winds."

@tool
def calculate(expression: str) -> str:
    """Perform mathematical calculations.

    Args:
        expression: Mathematical expression to evaluate (e.g., "2 + 2" or "sqrt(16)")
    """
    try:
        # Safe eval with limited scope
        import math
        allowed_names = {
            "abs": abs, "round": round, "min": min, "max": max,
            "sum": sum, "pow": pow, "sqrt": math.sqrt, "pi": math.pi,
            "sin": math.sin, "cos": math.cos, "tan": math.tan
        }
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return f"The result of {expression} is {result}"
    except Exception as e:
        return f"Error calculating: {str(e)}"

@tool
def search_database(query: str, filters: Optional[str] = None) -> str:
    """Search the database for information.

    Args:
        query: Search query string
        filters: Optional filters in JSON format
    """
    # Simulate database search
    return f"Found 3 results for '{query}' with filters: {filters or 'none'}"

@tool
def book_appointment(date: str, time: str, service: str) -> str:
    """Book an appointment for a service.

    Args:
        date: Date in YYYY-MM-DD format
        time: Time in HH:MM format
        service: Type of service to book
    """
    return f"Appointment booked for {service} on {date} at {time}"

@tool
def get_user_info(user_id: str) -> str:
    """Retrieve user information from the system.

    Args:
        user_id: The unique identifier for the user
    """
    # Simulate user lookup
    return f"User {user_id}: John Doe, Premium Member since 2024"

# Initialize LLM and tools
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Collect all tools
tools = [
    get_weather,
    calculate,
    search_database,
    book_appointment,
    get_user_info
]

# Create agent prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an intelligent assistant that identifies user intents and executes appropriate tools.

Available intents and their associated tools:
- weather_query: Use get_weather tool for weather-related questions
- calculation: Use calculate tool for math problems
- information_search: Use search_database tool for finding information
- appointment_booking: Use book_appointment tool for scheduling
- user_lookup: Use get_user_info tool for user information

Analyze the user's query, identify the most likely intent, and execute the appropriate tool(s) to fulfill their request.
Be direct and efficient in your tool usage."""),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# Create agent
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=3
)

# Intent classification function
def classify_intent(query: str) -> IntentMatch:
    """
    Classify the query into one of the predefined intents.
    Uses LLM to determine the most likely intent.
    """
    classification_prompt = f"""Classify this query into one of these intents:
- weather_query
- calculation
- information_search
- appointment_booking
- user_lookup

Query: "{query}"

Respond with just the intent name and a confidence score (0-1), formatted as:
intent: <name>
confidence: <score>"""

    response = llm.invoke([HumanMessage(content=classification_prompt)])
    response_text = response.content.strip()

    # Parse response
    lines = response_text.split('\n')
    intent = "unknown"
    confidence = 0.5

    for line in lines:
        if line.startswith("intent:"):
            intent = line.split(":", 1)[1].strip()
        elif line.startswith("confidence:"):
            try:
                confidence = float(line.split(":", 1)[1].strip())
            except:
                confidence = 0.5

    return IntentMatch(
        intent=intent,
        confidence=confidence,
        parameters={}
    )

# API Endpoints
@app.get("/")
async def root():
    """Root endpoint with API information"""
    return {
        "name": "Intent-Based Tool Execution API",
        "version": "1.0.0",
        "endpoints": {
            "/query": "POST - Execute query with intent matching",
            "/intents": "GET - List available intents",
            "/tools": "GET - List available tools"
        }
    }

@app.post("/query", response_model=QueryResponse)
async def process_query(request: QueryRequest):
    """
    Process a user query by:
    1. Classifying the intent
    2. Executing the appropriate tool(s)
    3. Returning the result
    """
    try:
        start_time = datetime.now()

        # Step 1: Classify intent
        intent_match = classify_intent(request.query)

        # Step 2: Execute agent with tools
        result = agent_executor.invoke({
            "input": request.query
        })

        # Calculate execution time
        execution_time = (datetime.now() - start_time).total_seconds()

        # Step 3: Return response
        return QueryResponse(
            query=request.query,
            intent_match=intent_match,
            result=result["output"],
            execution_time=execution_time,
            timestamp=datetime.now().isoformat()
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error processing query: {str(e)}")

@app.get("/intents")
async def list_intents():
    """List all available intents"""
    return {
        "intents": [
            {
                "name": "weather_query",
                "description": "Get weather information for a location",
                "example": "What's the weather in New York?"
            },
            {
                "name": "calculation",
                "description": "Perform mathematical calculations",
                "example": "Calculate 25 * 4 + 10"
            },
            {
                "name": "information_search",
                "description": "Search for information in the database",
                "example": "Find all orders from last month"
            },
            {
                "name": "appointment_booking",
                "description": "Book appointments for services",
                "example": "Book a haircut appointment for tomorrow at 3pm"
            },
            {
                "name": "user_lookup",
                "description": "Get user information",
                "example": "Show me details for user 12345"
            }
        ]
    }

@app.get("/tools")
async def list_tools():
    """List all available tools"""
    return {
        "tools": [
            {
                "name": tool.name,
                "description": tool.description,
                "args": tool.args
            }
            for tool in tools
        ]
    }

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "timestamp": datetime.now().isoformat()}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)