LangChain 工具注册与调用完全指南：@tool 装饰器到 Agent 实战

版本：Python 3.9+、LangChain ≥ 0.1
目标：学会「定义 → 绑定 → 调用 → 自动化」四种玩法，复制即可运行

---

0. 一键环境

pip install -U langchain langchain-openai python-dotenv

.env 同目录

OPENAI_API_KEY=sk-xxx

# 0_env.py
import os, dotenv
from langchain_openai import ChatOpenAI
dotenv.load_dotenv()
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

---

1. 最快上手：@tool 装饰器一行变工具

文件：1_tool_decorator.py

from langchain_core.tools import tool

@tool
def add(a: int, b: int) -> int:
    """计算两个整数之和"""
    return a + b

# 自动生成的 OpenAI 函数 JSON
print(add.name)        # add
print(add.description) # 计算两个整数之和
print(add.args)        # {'a': {'type': 'integer'}, 'b': {'type': 'integer'}}

# 直接运行
print(add.invoke({"a": 10, "b": 20}))   # 30

---

2. 高级定制：改名、加 schema、return_direct

文件：2_tool_custom.py

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

class AddInput(BaseModel):
    a: int = Field(description="被加数")
    b: int = Field(description="加数")

@tool(name="add_two_number", args_schema=AddInput, return_direct=True)
def add(a: int, b: int) -> int:
    """计算两个整数之和"""
    return a + b

print(add.args)  # 带字段描述的结构化 schema

• return_direct=True → 工具结果直接返回用户，不再经过 LLM 润色
• 字段描述会自动写进 OpenAI function JSON，提高模型传参准确率

---

3. 不用装饰器：StructuredTool.from_function

文件：3_structured_tool.py

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

class AddInput(BaseModel):
    a: int = Field(description="被加数")
    b: int = Field(description="加数")

def add(a: int, b: int) -> int:
    return a + b

tool = StructuredTool.from_function(
    func=add,
    name="add_two_number",
    description="计算两个整数之和",
    args_schema=AddInput,
    return_direct=True,
)
print(tool.invoke({"a": 5, "b": 7}))   # 12

适用：函数来自第三方库，无法修改源码加装饰器

---

4. 手动调用模型：functions 参数（无 Agent）

文件：4_manual_call.py

from langchain_core.messages import HumanMessage
from langchain_core.utils.function_calling import convert_to_openai_function
from 0_env import llm
from langchain.tools import MoveFileTool

tools = [MoveFileTool()]
functions = [convert_to_openai_function(t) for t in tools]

msg = llm.invoke(
    [HumanMessage(content="把 foo.txt 移动到 /tmp")],
    functions=functions
)

if msg.additional_kwargs.get("function_call"):
    name = msg.additional_kwargs["function_call"]["name"]
    args = msg.additional_kwargs["function_call"]["arguments"]
    print("即将调用：", name, args)
    # 执行工具
    result = tools[0].run(args)
    print("结果：", result)

---

5. 全自动链：bind_tools + LCEL（推荐）

文件：5_bind_tools.py

from langchain_core.prompts import ChatPromptTemplate
from 0_env import llm
from langchain.tools import MoveFileTool

tools = [MoveFileTool()]

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是文件助手，尽量简洁"),
    ("human", "{question}"),
])

# ① 绑定工具
llm_with_tools = llm.bind_tools(tools)

# ② 一条链搞定
chain = prompt | llm_with_tools

msg = chain.invoke({"question": "把 report.docx 移动到桌面"})
print(msg.additional_kwargs["function_call"])

---

6. 返回结果再交给模型：两步链

文件：6_two_step.py

from langchain_core.runnables import RunnablePassthrough

def call_tool(msg):
    if msg.additional_kwargs.get("function_call"):
        tool = MoveFileTool()
        args = msg.additional_kwargs["function_call"]["arguments"]
        return tool.run(args)
    return msg.content

full_chain = (
    prompt
    | llm.bind_tools([MoveFileTool()])
    | RunnablePassthrough(call_tool)   # 先执行工具
    | llm                              # 再把结果给模型润色
)

print(full_chain.invoke({"question": "移动文件后给我一句话总结"}))

---

7. 与 Agent 集成：create_openai_tools_agent（一行 Agent）

文件：7_agent.py

from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

system_prompt = "你是文件管家，帮助用户管理磁盘文件"
prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{input}"),
])

agent = create_openai_tools_agent(llm, [MoveFileTool()], prompt)
agent_executor = AgentExecutor(agent=agent, tools=[MoveFileTool()], verbose=True)

agent_executor.invoke({"input": "把 notes.txt 移动到 D:\\backup"})

---

8. 异步 & 流式：同样一条链

# 异步
result = await agent_executor.ainvoke({"input": "移动文件"})

# 流式
async for chunk in agent_executor.astream({"input": "移动文件"}):
    print(chunk, flush=True)

---

9. 常见坑速查

现象	原因
模型不传参	字段缺少 Field(description=...)
工具被调用却报错	参数 JSON 格式不对 → 用 Pydantic 校验
return_direct=True 无效	仅在 Agent 外手动调用时生效；Agent 会再走模型

---

10. 一张图总结选型

┌──────────────┐     ┌────────────┐     ┌──────────────────────┐
│    定义      │     │   绑定      │     │        调用          │
│   @tool      │────▶│ bind_tools │────▶│手动 / Agent         │
│StructuredTool│     │            │     │  流式 / 异步         │
└──────────────┘     └────────────┘     └──────────────────────┘

复制代码即可运行，把函数变成大模型“会调用的工具”，今天就到这里，下篇见！