🛠️ 工具体系

class BaseTool(RunnableSerializable[str | dict | ToolCall, Any]):
    """所有工具的基类。

    注意：它继承 RunnableSerializable —— 工具本身就是一个 Runnable！
    输入可以是 str/dict/ToolCall，输出任意。
    """

    name: str
    """工具名（模型在 tool_calls 里引用的就是它）"""

    description: str
    """工具描述 —— 极其重要！模型靠它判断"什么时候该用这个工具"。
    写得好坏直接决定 Agent 能不能正确选工具。"""

    args_schema: type[BaseModel]
    """参数 schema（Pydantic 模型），自动转成 JSON Schema 给模型"""

    @property
    def tool_call_schema(self) -> ArgsSchema:
        """返回给模型看的参数 schema。"""
        ...   # base.py:671

    def _run(self, *args, **kwargs) -> Any:
        """真实执行逻辑 —— 子类必须实现（同步版）。"""
        raise NotImplementedError

    async def _arun(self, *args, **kwargs) -> Any:
        """异步版执行逻辑（可选，默认降级到 _run）。"""

    def invoke(self, input, config=None, **kwargs):
        """统一调用入口（继承自 Runnable）。input 可以是 dict/ToolCall。
        内部会校验参数，然后调 _run。"""
        ...   # base.py:731

def tool(
    *,
    description: str | None = None,
    return_direct: bool = False,
    args_schema: ArgsSchema | None = None,
    infer_schema: bool = True,           # 自动从类型注解推断 schema
    ...
) -> Callable[..., BaseTool]:
    """把普通函数/Runnable 转成 BaseTool。"""

# ============ 用法 ============
from langchain_core.tools import tool

@tool
def search_weather(city: str) -> str:
    """查询指定城市的当前天气。

    Args:
        city: 城市名称，如"北京"、"上海"
    """
    # 这里的 docstring 会变成工具的 description！
    # city: str 的类型注解会变成 args_schema
    # 实际查询逻辑（这里假装）
    db = {"北京": "晴 25度", "上海": "多云 30度"}
    return db.get(city, "未知城市")

# 现在 search_weather 是一个 BaseTool 对象
print(search_weather.name)         # → "search_weather"
print(search_weather.description)  # → docstring 内容
print(search_weather.args_schema.model_json_schema())  # → 参数的 JSON Schema

# 直接调用（也可以用 .invoke）
print(search_weather.invoke({"city": "北京"}))  # → "晴 25度"

from langchain_core.tools import tool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    max_results: int = Field(default=5, ge=1, le=20, description="最大返回数 1-20")

@tool(args_schema=SearchInput)
def web_search(query: str, max_results: int = 5) -> str:
    """在网络上搜索信息。当需要查找最新资讯或模型不知道的事实时使用。"""
    return f"搜索 {query}，返回 {max_results} 条结果"

# 模型会看到带约束的 schema：max_results 必须在 1-20 之间
print(web_search.args_schema.model_json_schema())
# 调用时若参数非法，会抛错（而不是传给函数）

@tool(response_format="content_and_artifact")
def generate_chart(data: list[int]) -> tuple[str, bytes]:
    """生成柱状图。"""
    # 返回 (给模型看的文本, 附件如图片字节)
    return "图表已生成", b""

# invoke 时用 ToolMessage 取附件
result = generate_chart.invoke({"data": [1, 2, 3]})
# result.content = "图表已生成"，result.artifact = b""

from typing import Annotated
from langchain_core.tools import tool, InjectedToolArg

@tool
def get_user_files(
    pattern: str,                                    # 模型填这个
    user_id: Annotated[str, InjectedToolArg],        # 这个对模型隐藏，应用层注入
) -> str:
    """根据模式列出用户文件。"""
    return f"用户 {user_id} 的匹配 {pattern} 的文件"

# user_id 不会出现在给模型的 schema 里
# 但 invoke 时应用层必须提供
get_user_files.invoke({"pattern": "*.py", "user_id": "u123"})

# pip install langchain langchain-openai
# export OPENAI_API_KEY="sk-..."
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI

# ① 定义工具（@tool 自动生成 schema）
@tool
def add(a: int, b: int) -> int:
    """计算两个整数的和。当用户要求做加法时使用。"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """计算两个整数的乘积。当用户要求做乘法时使用。"""
    return a * b

tools = [add, multiply]

# ② 绑定到模型（LC3 学过的 bind_tools）
model = ChatOpenAI(model="gpt-4o-mini")
model_with_tools = model.bind_tools(tools)

# ③ 模型决定调用工具
ai_msg = model_with_tools.invoke("3乘以4等于多少？")
print("模型要调用:", ai_msg.tool_calls)
# → [{'name': 'multiply', 'args': {'a': 3, 'b': 4}, 'id': 'call_xxx'}]

# ④ 应用层执行工具（关键！模型不会自己执行）
tc = ai_msg.tool_calls[0]
tool_to_call = {"add": add, "multiply": multiply}[tc["name"]]
result = tool_to_call.invoke(tc["args"])
print("工具结果:", result)  # → 12

# ⑤ 把结果用 ToolMessage 喂回去（LC1 学过）
tool_msg = ToolMessage(content=str(result), tool_call_id=tc["id"])
final = model_with_tools.invoke([
    HumanMessage(content="3乘以4等于多少？"),
    ai_msg,
    tool_msg,
])
print("最终回答:", final.content)  # → "3乘以4等于12。"