Tool Calling：协议与封装

import json
from openai import OpenAI

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市当前的实时天气。当用户询问天气、温度、"
                           "降雨、穿衣建议等问题时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名，如「北京」",
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位，默认 celsius",
                    },
                },
                "required": ["city"],
            },
        },
    },
]

resp = client.chat.completions.create(
    model="qwen3:4b",
    messages=[{"role": "user", "content": "北京今天多少度？适合跑步吗"}],
    tools=tools,
)

msg = resp.choices[0].message
print(msg.content)        # 通常为空，或只有简短的思考文字
print(msg.tool_calls)      # 真正的货在这里

[ChoiceMessageToolCall(
    id='call_abc123',
    function=Function(
        name='get_weather',
        arguments='{"city": "北京"}'    # ← 一段 JSON 字符串
    )
)]

def get_weather(city: str, unit: str = "celsius") -> str:
    """演示用假实现；真实场景换成任意天气 API"""
    data = {"北京": 12, "上海": 18}
    t = data.get(city, 20)
    if unit == "fahrenheit":
        t = t * 9 / 5 + 32
    return f"{city}当前 {t}°{'C' if unit == 'celsius' else 'F'}，晴，空气质量优"

tc = msg.tool_calls[0]
args = json.loads(tc.function.arguments)          # JSON 字符串 → dict
result = get_weather(**args)                       # 真正的执行发生在这里
print(result)                                      # 北京当前 12°C，晴，空气质量优

messages = [
    {"role": "user", "content": "北京今天多少度？适合跑步吗"},
    msg,                                                    # 调用意向（assistant）
    {"role": "tool", "tool_call_id": tc.id, "content": result},  # 执行结果
]

final = client.chat.completions.create(
    model="qwen3:4b", messages=messages, tools=tools,
)
print(final.choices[0].message.content)
# 模型现在能看到工具结果，产出最终回答：
# "北京当前 12°C，晴，空气质量优，很适合跑步。不过温度略低，建议穿件薄外套。"

print(len(msg.tool_calls))    # 可能是 2：get_weather(北京) + get_weather(上海)

import concurrent.futures as cf

messages.append(msg)                                   # 先把意向入列
with cf.ThreadPoolExecutor() as pool:
    futures = {tc.id: pool.submit(
        get_weather, **json.loads(tc.function.arguments)
    ) for tc in msg.tool_calls}
    for tc in msg.tool_calls:
        messages.append({
            "role": "tool",
            "tool_call_id": tc.id,
            "content": str(futures[tc.id].result()),
        })

def chat_with_tools(user_input: str, tools: list, executor: dict,
                    max_rounds: int = 5) -> str:
    """executor: {工具名: 可调用函数} 的映射"""
    messages = [{"role": "user", "content": user_input}]
    for _ in range(max_rounds):
        resp = client.chat.completions.create(
            model="qwen3:4b", messages=messages, tools=tools,
        )
        msg = resp.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:                       # 不再调用 → 最终回答
            return msg.content

        for tc in msg.tool_calls:
            func = executor.get(tc.function.name)
            if func is None:                          # 防御：模型幻觉出工具名
                result = f"错误：不存在名为 {tc.function.name} 的工具"
            else:
                try:
                    result = str(func(**json.loads(tc.function.arguments)))
                except Exception as e:                # 防御：执行失败不崩溃
                    result = f"工具执行出错：{e}"
            messages.append({
                "role": "tool", "tool_call_id": tc.id, "content": result,
            })
    return "（达到最大轮次限制）"

# ❌ 只说"是什么"
"description": "获取天气"

# ✅ 说清"什么时候用、能干什么、边界在哪"
"description": "查询指定城市当前的实时天气（温度、天气现象、空气质量）。
当用户询问天气、气温、是否下雨下雪、该穿什么衣服等问题时使用。
仅支持中国主要城市。不提供未来 3 天以外的预报。"

def read_file(path: str) -> str:
    content = Path(path).read_text(encoding="utf-8")
    if len(content) > 4000:
        return content[:4000] + f"\n...(已截断，全文共 {len(content)} 字符)"
    return content

# ❌ 错误示例
"error"                        # 模型：？？然后开始编

# ✅ 可行动的错误
"城市 '北jing' 不存在。请检查拼写，或改用支持的城市列表中的名称"

import json
from pydantic import BaseModel, ValidationError

class Tool:
    """把一个 Python 函数封装成模型可用的工具"""

    def __init__(self, func, args_model: type[BaseModel], description: str):
        self.func = func
        self.args_model = args_model
        self.name = func.__name__
        self.description = description

    def schema(self) -> dict:
        """生成 OpenAI 格式的工具声明（本章第 2 节的 tools 列表元素）"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.args_model.model_json_schema(),
            },
        }

    def run(self, arguments_json: str) -> str:
        """执行工具。所有失败都以可读字符串返回，绝不抛异常"""
        try:
            args = self.args_model.model_validate_json(arguments_json)
            return str(self.func(**args.model_dump()))
        except ValidationError as e:
            return f"参数校验失败：{e}"
        except Exception as e:
            return f"工具执行出错：{type(e).__name__}: {e}"


class Registry:
    """工具注册器：schema 生成 + 名称分发"""

    def __init__(self):
        self.tools: dict[str, Tool] = {}

    def register(self, tool: Tool) -> None:
        self.tools[tool.name] = tool

    def schemas(self) -> list[dict]:
        return [t.schema() for t in self.tools.values()]

    def dispatch(self, name: str, arguments_json: str) -> str:
        tool = self.tools.get(name)
        if tool is None:
            return f"错误：不存在名为 {name} 的工具。可用工具：{list(self.tools)}"
        return tool.run(arguments_json)

import re
from pathlib import Path
from pydantic import BaseModel, Field, ValidationError

# ---------- read_file ----------
class ReadFileArgs(BaseModel):
    path: str = Field(description="要读取的文件路径（相对当前目录或绝对路径）")
    max_chars: int = Field(default=4000, description="最多返回的字符数")

def read_file(path: str, max_chars: int = 4000) -> str:
    p = Path(path)
    if not p.exists():
        return f"文件不存在：{path}"
    if not p.is_file():
        return f"路径不是文件：{path}"
    content = p.read_text(encoding="utf-8")
    if len(content) > max_chars:
        return content[:max_chars] + f"\n...(已截断，共 {len(content)} 字符)"
    return content

# ---------- write_file ----------
class WriteFileArgs(BaseModel):
    path: str = Field(description="目标文件路径")
    content: str = Field(description="要写入的完整内容（全量覆盖，非追加）")

def write_file(path: str, content: str) -> str:
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(content, encoding="utf-8")
    return f"已写入 {path}（{len(content)} 字符）"

# ---------- calculate ----------
class CalculateArgs(BaseModel):
    expression: str = Field(description="Python 算术表达式，如 '(12.5 + 3) * 2'")

def calculate(expression: str) -> str:
    # 只放行安全子集，杜绝 eval 直接执行任意代码
    if not re.fullmatch(r"[0-9+\-*/().%\s]+", expression):
        return "表达式包含不支持的字符（仅支持数字与 + - * / % ( )）"
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"计算失败：{e}"

# ---------- 注册 ----------
registry = Registry()
registry.register(Tool(read_file, ReadFileArgs,
    "读取本地文本文件的内容。文件过大时自动截断。"))
registry.register(Tool(write_file, WriteFileArgs,
    "将文本内容写入本地文件（覆盖写入）。需要创建或修改文件时使用。"))
registry.register(Tool(calculate, CalculateArgs,
    "精确计算算术表达式。任何数学计算都必须用本工具，不要心算。"))