第一个AI应用开发:带工具的聊天调用

import jsonimport httpximport osfrom dotenv import load_dotenvload_dotenv()ORDER_SERVICE_URL = os.getenv("ORDER_SERVICE_URL", "http://localhost:8080/api/orders")# 工具定义（符合OpenAI function calling格式）TOOLS = [    {        "type": "function",        "function": {            "name": "get_order_info",            "description": "查询指定订单号的详细信息，包括金额、状态、商品等",            "parameters": {                "type": "object",                "properties": {                    "order_id": {                        "type": "string",                        "description": "订单号，例如 ORD123"                    }                },                "required": ["order_id"],                "additionalProperties": False            }        }    },    {        "type": "function",        "function": {            "name": "get_weather",            "description": "查询某个城市的当前天气",            "parameters": {                "type": "object",                "properties": {                    "city": {                        "type": "string",                        "description": "城市名称，例如 北京"                    }                },                "required": ["city"],                "additionalProperties": False            }        }    }]async def execute_tool(tool_name: str, arguments: dict) -> str:    """执行工具并返回结果（字符串）"""    if tool_name == "get_order_info":        order_id = arguments.get("order_id")        if not order_id:            return "错误：缺少订单号"        # 调用Java订单服务        async with httpx.AsyncClient(timeout=10.0) as client:            try:                resp = await client.get(f"{ORDER_SERVICE_URL}/{order_id}")                if resp.status_code == 200:                    data = resp.json()                    return f"订单{order_id}：金额{data['amount']}元，状态{data['status']}，商品{data['product']}"                else:                    return f"未找到订单{order_id}"            except Exception as e:                return f"查询订单失败：{str(e)}"    elif tool_name == "get_weather":        city = arguments.get("city")        # 模拟天气API（可替换为真实API如wttr.in）        # 这里简单返回固定数据        weather_data = {            "北京": "晴，25°C",            "上海": "多云，28°C",            "深圳": "雷阵雨，30°C"        }        return weather_data.get(city, f"未知城市{city}，天气信息不可用")    else:        return f"未知工具：{tool_name}"

from fastapi import FastAPI, HTTPExceptionfrom fastapi.requests import Requestfrom fastapi.responses import JSONResponsefrom fastapi.responses import StreamingResponsefrom pydantic import BaseModelimport httpximport jsonfrom dotenv import load_dotenvfrom typing import List, Any, Dictfrom tools import TOOLS, execute_toolfrom config import ConfigParamclass Message(BaseModel):    role: str  # user, assistant, tool    content: str = None    tool_calls: Any = None    tool_call_id: str = Noneclass ChatARequest(BaseModel):    messages: List[Message]  # 对话历史    # 可以添加temperature等参数class ChatAResponse(BaseModel):    reply: str    messages: List[Message]  # 更新后的对话历史（包含assistant回复及可能的tool消息）# ---------- OpenAI 后端 ----------async def chat_with_openai(messages: List[Dict], tools=None):    from openai import AsyncOpenAI    client = AsyncOpenAI(api_key=ConfigParam.MODEL_BASE_URL)    response = await client.chat.completions.create(        model=ConfigParam.LLM_MODEL,        messages=messages,        tools=tools,        tool_choice="auto",        temperature=0.2    )    return response# ---------- Ollama 后端 ----------async def chat_with_ollama(messages: List[Dict], tools=None):    # Ollama 支持 tools 参数，格式与 OpenAI 类似    baseUrl = ConfigParam.MODEL_BASE_URL    if baseUrl.endswith('/'):        baseUrl = baseUrl[:-1]    url = f"{baseUrl}/api/chat"    payload = {        "model": ConfigParam.LLM_MODEL,        "messages": messages,        "stream": False,        "options": {"temperature": 0.2}    }    if tools:        payload["tools"] = tools  # Ollama 0.3+ 支持    async with httpx.AsyncClient(timeout=60.0) as client:        print("Ollama request:", payload)        resp = await client.post(url, json=payload)        resp.raise_for_status()        data = resp.json()        print("Ollama response:", data)        # 构造类似 OpenAI 的响应对象        class MockChoice:            class MockMessage:                def __init__(self, d):                    self.content = d.get("content")                    self.tool_calls = d.get("tool_calls")            def __init__(self, d):                self.message = self.MockMessage(d["message"])        class MockResponse:            def __init__(self, d):                self.choices = [MockChoice(d)]        return MockResponse(data)# 选择后端LLM_BACKEND = "ollama"# ---------- 统一处理循环 ----------async def run_conversation(messages: List[Dict]) -> List[Dict]:    """处理tool calling循环，返回最终的messages列表（包含assistant最终回复）"""    # 复制一份，避免修改原列表    conversation = messages.copy()    max_iter = 5  # 防止无限循环    for _ in range(max_iter):        # 调用LLM        if LLM_BACKEND == "openai":            response = await chat_with_openai(conversation, tools=TOOLS)        else:            response = await chat_with_ollama(conversation, tools=TOOLS)        assistant_msg = response.choices[0].message        # 将assistant消息加入对话        assistant_dict = assistant_msg.model_dump() if LLM_BACKEND == "openai" else {            "role": "assistant",            "content": assistant_msg.content,            "tool_calls": assistant_msg.tool_calls        }        conversation.append(assistant_dict)        # 如果没有tool_calls，结束循环        if not assistant_msg.tool_calls:            break        # 有tool_calls，执行每个工具并追加tool消息        for tool_call in assistant_msg.tool_calls:            print(f"Received tool call: {tool_call}")            tool_call = tool_call if isinstance(tool_call, dict) else tool_call.model_dump()            tool_name = tool_call["function"]["name"]            print(f"Received tool call: {tool_name}")            arg = tool_call["function"]["arguments"]            if isinstance(arg, str):                arguments = json.loads(arg)            else:                arguments = arg            tool_result = await execute_tool(tool_name, arguments)            print(f"Executed tool {tool_name} with args {arguments}, got result: {tool_result}")            # 添加tool response消息            conversation.append({                "role": "tool",                "tool_call_id": tool_call["id"],                "content": tool_result            })    return conversation@app.post("/chat_tool", response_model=ChatAResponse)async def chat(request: ChatARequest):    # 将pydantic模型转为dict列表    messages = [msg.model_dump(exclude_none=True) for msg in request.messages]    try:        updated_messages = await run_conversation(messages)        # 提取最后一条assistant消息的content作为回复        last_assistant = None        for msg in reversed(updated_messages):            if msg.get("role") == "assistant" and msg.get("content"):                last_assistant = msg["content"]                break        # 将updated_messages转回Message列表（简化，不要求完整还原tool_calls结构）        # 直接返回字符串和完整历史（前端可以存储）        return ChatAResponse(            reply=last_assistant or "抱歉，没有生成回复。",            messages=[Message(**m) for m in updated_messages if m.get("role") != "tool" or True]  # 保留tool消息        )    except Exception as e:        print("Error in /chat_tool:", str(e))        raise HTTPException(status_code=500, detail=str(e))if __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="0.0.0.0", port=8000)