AI 工具全栈实践：从模型接入到工具调用的工程化实现

pip install fastapi uvicorn openai pydantic

import os, json, ast
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)

app = FastAPI(title="AI Tool Fullstack Demo")

class ChatRequest(BaseModel):
    message: str

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询城市天气，仅用于演示",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "计算简单数学表达式",
            "parameters": {
                "type": "object",
                "properties": {"expression": {"type": "string"}},
                "required": ["expression"],
            },
        },
    },
]

def get_weather(city: str) -> dict:
    # 真实项目应调用天气 API，并做缓存、限流、错误处理
    return {"city": city, "weather": "晴", "temperature": 26}

def calculate(expression: str) -> dict:
    node = ast.parse(expression, mode="eval")
    allowed = (
        ast.Expression, ast.BinOp, ast.UnaryOp, ast.Constant,
        ast.Add, ast.Sub, ast.Mult, ast.Div, ast.USub, ast.UAdd,
    )
    for n in ast.walk(node):
        if not isinstance(n, allowed):
            raise ValueError("表达式包含不允许的语法")
    result = eval(compile(node, "<calc>", "eval"), {"__builtins__": {}}, {})
    return {"result": result}

TOOL_MAP = {"get_weather": get_weather, "calculate": calculate}

@app.post("/chat")
def chat(req: ChatRequest):
    messages = [
        {"role": "system", "content": "你是助手。需要时调用工具，不要编造工具结果。"},
        {"role": "user", "content": req.message},
    ]

    for _ in range(5):
        resp = client.chat.completions.create(
            model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )
        msg = resp.choices[0].message
        messages.append(msg.model_dump(exclude_none=True))

        if not msg.tool_calls:
            return {"reply": msg.content}

        for call in msg.tool_calls:
            fn = TOOL_MAP.get(call.function.name)
            try:
                args = json.loads(call.function.arguments)
                result = fn(**args) if fn else {"error": "unknown tool"}
            except Exception as e:
                result = {"error": str(e)}

            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": json.dumps(result, ensure_ascii=False),
            })

    raise HTTPException(status_code=500, detail="工具调用轮次超限")

uvicorn main:app --reload

const res = await fetch("/chat", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ message: "北京天气如何？再算一下 12*(3+4)" })
});
console.log(await res.json());