工具系统#

1
# pip install langchain langchain-openai
2

3
from langchain_core.tools import tool
4

5
@tool
6
def multiply(a: int, b: int) -> int:
7
    """将两个整数相乘。当用户需要计算乘法时使用此工具。
8

9
    Args:
10
        a: 第一个整数
11
        b: 第二个整数
12
    """
13
    return a * b
14

15
# 查看工具的元信息
16
print(multiply.name)          # "multiply"
17
print(multiply.description)   # "将两个整数相乘。当用户需要计算乘法时使用此工具。"
18
print(multiply.args_schema.model_json_schema())
19
# {'properties': {'a': {'title': 'A', 'type': 'integer'},
20
#                 'b': {'title': 'B', 'type': 'integer'}},
21
#  'required': ['a', 'b'],
22
#  'title': 'multiply',
23
#  'type': 'object'}

1
# pip install langchain pydantic
2

3
from langchain_core.tools import StructuredTool
4
from pydantic import BaseModel, Field
5

6
class SearchInput(BaseModel):
7
    """搜索工具的输入参数。"""
8
    query: str = Field(description="搜索关键词")
9
    max_results: int = Field(default=5, description="最大返回结果数", ge=1, le=20)
10
    language: str = Field(default="zh", description="搜索语言，如 'zh'、'en'")
11

12
def search_web(query: str, max_results: int = 5, language: str = "zh") -> str:
13
    """搜索互联网获取实时信息。当用户询问最新事件、实时数据时使用。"""
14
    # 这里替换为实际的搜索 API 调用
15
    return f"搜索 '{query}' 的前 {max_results} 条结果（{language}）..."
16

17
search_tool = StructuredTool.from_function(
18
    func=search_web,
19
    name="search_web",
20
    description="搜索互联网获取实时信息。当用户询问最新事件、实时数据时使用。",
21
    args_schema=SearchInput,
22
)
23

24
print(search_tool.name)        # "search_web"
25
print(search_tool.args_schema.model_json_schema())
26
# 输出包含 query、max_results、language 的完整 JSON Schema

1
# pip install langchain-community tavily-python
2

3
from langchain_community.tools.tavily_search import TavilySearchResults
4

5
# 需要设置环境变量 TAVILY_API_KEY
6
search = TavilySearchResults(max_results=3)
7
results = search.invoke("LangChain 最新版本")
8
print(results)

1
# pip install langchain aiohttp
2

3
import aiohttp
4
from langchain_core.tools import tool
5

6
@tool
7
async def async_search(query: str) -> str:
8
    """异步搜索工具，用于高并发场景下的互联网搜索。"""
9
    async with aiohttp.ClientSession() as session:
10
        async with session.get(
11
            "https://api.example.com/search",
12
            params={"q": query}
13
        ) as resp:
14
            data = await resp.json()
15
            return str(data["results"][:3])
16

17
# 异步调用
18
# result = await async_search.ainvoke({"query": "LangChain Agent"})