Function Calling：模型提参数，应用完成受控执行闭环

python -m pip install "langchain>=1,<2" "langchain-openai>=1,<2" "pydantic>=2,<3"

from typing import Literal
from pydantic import BaseModel, Field, ValidationError
from langchain_core.messages import HumanMessage, ToolMessage


class DeviceQuery(BaseModel):
    device_id: str = Field(pattern=r"^AX-\d{1,4}$")
    field: Literal["status", "temperature"]


def query_device(device_id: str, field: str) -> dict:
    """查询指定设备的状态或温度。"""
    database = {"AX-3": {"status": "online", "temperature": 41.2}}
    if device_id not in database:
        return {"found": False}
    return {"found": True, "device_id": device_id, field: database[device_id][field]}


tools = {"query_device": (DeviceQuery, query_device)}
llm_with_tools = model.bind_tools([query_device])
messages = [HumanMessage("AX-3 当前温度是多少？")]
response = llm_with_tools.invoke(messages)
messages.append(response)

for call in response.tool_calls:
    name, args, call_id = call["name"], call["args"], call["id"]
    if name not in tools:
        output = {"ok": False, "error": "tool_not_allowed"}
    else:
        schema, executor = tools[name]
        try:
            validated = schema.model_validate(args)
            output = {"ok": True, "data": executor(**validated.model_dump())}
        except ValidationError as error:
            output = {"ok": False, "error": "invalid_arguments", "detail": error.errors()}
    messages.append(ToolMessage(str(output), tool_call_id=call_id))

final = llm_with_tools.invoke(messages)
print(final.text)