Secure ، Swift و Smart: یک راهنمای اصلی برای ساخت گردش کار عامل AI با مدل های محلی

SYSTEM_INSTRUCTION = """
You are a MS SQL expert Db Admin agent. Your job is to assist users in querying a Microsoft SQL Server database using provided tools. Address users casually and:

1. Identify database and table names, even from vague inputs.
2. Use tools to resolve ambiguities before asking for clarification.
3. Handle SQL operations: SELECTs, schema inspection, stored procedures, and more.
4. Default to 25 records for general SELECTs unless specified.
5. Format results clearly: tables for data, SQL for schemas.
6. Only seek clarification when truly stuck.

Available tools:
- list_databases: List all databases.
- all_table_names: List tables in a database.
- filter_table_names: Find tables matching a substring.
- schema_definitions: Get table schemas.
- execute_query: Run a SQL query.
"""

from fastapi import FastAPI
from mcp_agent.app import MCPApp
from mcp_agent.agents.agent import Agent

app = FastAPI(title="SQL Agent API", version="2.0")
mcp_app = MCPApp(name="sql_agent")

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    body = await request.json()
    user_prompt = next((m["content"] for m in reversed(body.get("messages", [])) if m["role"] == "user"), "")
    async def stream_response():
        async for chunk in sql_agent_workflow(user_prompt):
            yield chunk
    return StreamingResponse(content=stream_response(), media_type="text/event-stream")

def classify_query(prompt):
    prompt_lower = prompt.lower()
    if re.search(r"execute\s+this\s+query[^:]*:", prompt_lower):
        return "execute_query"
    if "schema" in prompt_lower and "table" in prompt_lower:
        return "table_schema"
    return "general_query"

def extract_database_name(prompt):
    prompt_lower = prompt.lower()
    db_aliases = {"enterprisedb": "enterprise_db", "timesheets": "timesheets_db"}
    explicit_match = re.search(r"(?:in|on|for|from)\s+([a-z0-9_\s.-]+?)(?:\s+database|\s+db)?$", prompt_lower)
    if explicit_match:
        db_name = explicit_match.group(1).strip()
        return db_aliases.get(db_name, db_name)
    return "master"  # Default fallback

TOOL_RESULT_CACHE: Dict[str, Dict[str, Any]] = {}
DEFAULT_CACHE_TTL_SECONDS = 600

async def call_tool_with_cache(sql_agent: Agent, tool_name: str, tool_args: Dict[str, Any], logger, ttl_seconds: int = DEFAULT_CACHE_TTL_SECONDS):
    cache_key = f"{tool_name}::{json.dumps(tool_args, sort_keys=True)}"
    current_time = time.time()
    if cache_key in TOOL_RESULT_CACHE and (current_time - TOOL_RESULT_CACHE[cache_key]["timestamp"]) < ttl_seconds:
        logger.info(f"Cache HIT for '{tool_name}'")
        return TOOL_RESULT_CACHE[cache_key]["result_text"]
    result = await sql_agent.call_tool(tool_name, tool_args)
    result_text = result.content[0].text if result.content else ""
    TOOL_RESULT_CACHE[cache_key] = {"timestamp": current_time, "result_text": result_text}
    return result_text

async def sql_agent_workflow(user_prompt: str) -> AsyncGenerator[str, None]:
    async with mcp_app.run() as agent_app:
        sql_agent = Agent(name="sql_agent", instruction=SYSTEM_INSTRUCTION, server_names=["sql_server"])
        async with sql_agent:
            llm = await sql_agent.attach_llm(OpenAIAugmentedLLM)  # Local 14B model
            query_type = classify_query(user_prompt)
            db_name = extract_database_name(user_prompt)

            if query_type == "execute_query":
                query = re.search(r"execute\s+this\s+query[^:]*:\s*([^;]+);", user_prompt, re.IGNORECASE).group(1)
                result = await call_tool_with_cache(sql_agent, "execute_query", {"query": query, "database": db_name}, agent_app.logger)
                yield f"Dude, here’s your result:\n{result}"
                return

            max_iterations = 25
            for _ in range(max_iterations):
                response = await llm.generate_str(message=user_prompt, request_params=RequestParams(model="local_14b"))
                if "" in response:
                    yield "Hey, I’m stuck! Can you clarify: " + response.split("")[1].split("")[0]
                    return
                tool_name, tool_args = await extract_tool_call(response)
                if tool_name:
                    result = await call_tool_with_cache(sql_agent, tool_name, tool_args, agent_app.logger)
                    yield f"Tool {tool_name} result:\n{result}"
                    if tool_name == "execute_query":
                        return
                user_prompt = f"Based on: {result}, continue: {user_prompt}"