1
from pydantic import BaseModel, Field
2

3
from langchain.tools import tool
4
from langchain.messages import HumanMessage
5
from langchain_core.utils.function_calling import convert_to_openai_tool
6

7
class WeatherSchema(BaseModel):
8
    city: str = Field(default="北京", description="城市名称")
9
    if_forecast: bool = Field(default=False, description="是否包含明日天气预报")
10

11
# 工具名、description、args_schema 三样一起给
12
@tool("get_weather_and_forecast", description="查询当日天气，可以包含明日天气预报", args_schema=WeatherSchema)
13
def get_weather(city: str, if_forecast: bool):
14
    res = f"{city} 今天天气不错"
15
    if if_forecast:
16
        res += "\n明天也不错"
17
    return res
18

19
print(convert_to_openai_tool(get_weather))
20

21
model_with_tools = model.bind_tools([get_weather])
22

23
messages = [HumanMessage("今天杭州天气如何？明天呢？")]
24
response = model_with_tools.invoke(messages)
25
messages.append(response)
26

27
for tool_call in response.tool_calls:
28
    if tool_call["name"] == "get_weather_and_forecast":
29
        tool_msg = get_weather.invoke(tool_call)     # 注意：tool_call 能直接进 invoke
30
        messages.append(tool_msg)
31

32
final_response = model_with_tools.invoke(messages)
33
messages.append(final_response)
34

35
for msg in messages:
36
    msg.pretty_print()

1
================================ Human Message =================================
2

3
今天杭州天气如何？明天呢？
4
================================== Ai Message ==================================
5
Tool Calls:
6
  get_weather_and_forecast (call_c82UrCpAdVAqiHfl4OkZocbg)
7
 Call ID: call_c82UrCpAdVAqiHfl4OkZocbg
8
  Args:
9
    city: 杭州
10
    if_forecast: True
11
================================= Tool Message =================================
12
Name: get_weather_and_forecast
13

14
杭州 今天天气不错
15
明天也不错
16
================================== Ai Message ==================================
17

18
杭州今天：天气不错。
19
杭州明天：也不错。

1
from langchain.tools import tool
2
from langchain.messages import HumanMessage
3
from langchain_core.utils.function_calling import convert_to_openai_tool
4

5
@tool("get_weather_and_forecast", parse_docstring=True)
6
def get_weather(city: str = "北京", if_forecast: bool = False):
7
    """
8
    查询当日天气，可以包含明日天气预报
9

10
    Args:
11
        city: 城市名称
12
        if_forecast: 是否包含明日天气预报
13
    """
14
    res = f"{city} 今天天气不错"
15
    if if_forecast:
16
        res += "\n明天要下雨"
17
    return res
18

19
print(convert_to_openai_tool(get_weather))
20

21
model_with_tools = model.bind_tools([get_weather])
22

23
messages = [HumanMessage("今天杭州天气如何？明天呢？")]
24
response = model_with_tools.invoke(messages)
25
messages.append(response)
26

27
# 将工具调用的结果添加到消息列表中
28
for tool_call in response.tool_calls:
29
    if tool_call["name"] == "get_weather_and_forecast":
30
        # 返回值 tool_msg 类型是 ToolMessage
31
        tool_msg = get_weather.invoke(tool_call)
32
        messages.append(tool_msg)
33

34
final_response = model_with_tools.invoke(messages)
35
messages.append(final_response)
36

37
for msg in messages:
38
    msg.pretty_print()

1
from langchain_core.tools import tool
2
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
3

4
# 1. 定义股票查询工具
5
@tool(parse_docstring=True)
6
def get_stock_price(company: str, timeframe: str = "today") -> str:
7
    """获取指定公司的股票价格信息
8

9
    Args:
10
        company: 公司名称（如：苹果公司, 微软公司, 谷歌公司）
11
        timeframe: 时间范围（today-今日, week-本周, month-本月）
12
    """
13
    # 模拟股票数据
14
    mock_data = {
15
        "苹果公司": {"today": 185.20, "week": 183.50, "month": 180.75},
16
        "微软公司": {"today": 415.86, "week": 412.30, "month": 405.42},
17
        "谷歌公司": {"today": 15.42, "week": 15.20, "month": 14.85},
18
    }
19
    if company in mock_data:
20
        price = mock_data[company].get(timeframe, "未知时间范围")
21
        return f"{company} {timeframe}价格: {price}美元"
22
    else:
23
        return f"未找到股票代码 {company} 的数据"
24

25
# 定义新闻搜索工具
26
@tool(parse_docstring=True)
27
def search_news(company: str) -> str:
28
    """搜索指定公司的财经新闻
29

30
    Args:
31
        company: 公司名称
32

33
    Returns:
34
        公司的财经新闻，每个新闻占一行
35
    """
36
    mock_news = {
37
        "苹果公司": [
38
            "苹果发布新款iPhone，股价上涨3%",
39
            "苹果与欧盟达成反垄断和解协议",
40
            "苹果将在印度扩大生产规模",
41
        ],
42
        "微软公司": [
43
            "微软Azure云业务季度增长超预期",
44
            "微软完成对Nuance的收购",
45
            "微软推出新一代AI助手Copilot",
46
        ],
47
    }
48
    news_list = mock_news.get(company, [f"未找到{company}的相关新闻"])
49
    return "\n".join(news_list)
50

51
# 2. 初始化模型并绑定工具
52
tools = [get_stock_price, search_news]
53
model_with_tools = model.bind_tools(tools)
54

55
message_list = []
56
human_message = HumanMessage(content="苹果公司今天的股价是多少？最近有什么新闻？")
57
# human_message = HumanMessage(content="比较一下微软和苹果的股价")
58
# human_message = HumanMessage(content="腾讯最近有什么重大新闻？")
59
# human_message = HumanMessage(content="海水为什么是咸的？")
60
message_list.append(human_message)
61

62
# 3. 工具调用循环
63
while True:
64
    response = model_with_tools.invoke(message_list)
65
    message_list.append(response)
66

67
    # 如果模型不需要调用工具，直接退出循环
68
    if not response.tool_calls:
69
        print("没有工具调用，直接返回答案")
70
        break
71

72
    # 4. 开发者根据模型的响应，调用工具并获取结果
73
    for tool_call in response.tool_calls:
74
        if tool_call["name"] == "get_stock_price":
75
            stock_result = get_stock_price.invoke(tool_call)
76
            print("stock_result", stock_result)
77
            message_list.append(stock_result)
78
        if tool_call["name"] == "search_news":
79
            news_result = search_news.invoke(tool_call)
80
            print("news_result", news_result)
81
            message_list.append(news_result)
82

83
for msg in message_list:
84
    msg.pretty_print()

1
================================ Human Message =================================
2

3
苹果公司今天的股价是多少？最近有什么新闻？
4
================================== Ai Message ==================================
5
Tool Calls:
6
  get_stock_price (call_cpGOhWce8rlIFSZ2G9w7ouON)
7
 Call ID: call_cpGOhWce8rlIFSZ2G9w7ouON
8
  Args:
9
    company: 苹果公司
10
    timeframe: today
11
  search_news (call_W9zjkAO8TNCmUbed2ld9tsxl)
12
 Call ID: call_W9zjkAO8TNCmUbed2ld9tsxl
13
  Args:
14
    company: 苹果公司
15
================================= Tool Message =================================
16
Name: get_stock_price
17

18
苹果公司 today价格: 185.2美元
19
================================= Tool Message =================================
20
Name: search_news
21

22
苹果发布新款iPhone，股价上涨3%
23
苹果与欧盟达成反垄断和解协议
24
苹果将在印度扩大生产规模
25
================================== Ai Message ==================================
26

27
苹果公司今天股价是 **185.2 美元**。
28

29
最近新闻包括：
30
- 苹果发布了**新款 iPhone**，带动股价上涨约 **3%**
31
- 苹果与欧盟达成了**反垄断和解协议**
32
- 苹果计划在**印度扩大生产规模**

1
from langchain.tools import tool
2
from langchain.messages import HumanMessage
3

4
@tool(parse_docstring=True)
5
def get_weather(city: str) -> str:
6
    """
7
    获取当日天气
8

9
    Args:
10
        city: 城市名称
11
    """
12
    return f'{city}当天晴朗'
13

14
@tool(parse_docstring=True)
15
def get_news() -> str:
16
    """
17
    获取当日新闻
18
    """
19
    return "近期，受全球芯片短缺等多重因素影响，多地回收商称废旧手机回收市场迎来“火热潮”，回收价格普遍上涨，旧手机成“香饽饽”。"
20

21
model_with_tools = model.bind_tools([get_weather, get_news])
22

23
messages = [
24
    HumanMessage("今天杭州天气如何？今天新闻是什么？别瞎编")
25
]
26

27
response = model_with_tools.invoke(messages)
28
response.pretty_print()

1
messages.append(response)
2

3
for tool_call in response.tool_calls:
4
    if tool_call["name"] == "get_weather":
5
        tool_msg = get_weather.invoke(tool_call)
6
        print(tool_msg)
7
        messages.append(tool_msg)
8
    elif tool_call["name"] == "get_news":
9
        tool_msg = get_news.invoke(tool_call)
10
        print(tool_msg)
11
        messages.append(tool_msg)
12
    else:
13
        raise Exception("不存在的工具")
14

15
final_response = model.invoke(messages)
16
messages.append(final_response)
17

18
for msg in messages:
19
    msg.pretty_print()

1
from langchain.tools import tool
2
from langchain.messages import HumanMessage
3

4
@tool(parse_docstring=True)
5
def get_weather1(city: str) -> str:
6
    """
7
    获取当日天气
8

9
    Args:
10
        city: 城市名称
11
    """
12
    return f'{city}当天晴朗'
13

14
@tool(parse_docstring=True)
15
def get_weather2(city: str) -> str:
16
    """
17
    获取当日天气
18

19
    Args:
20
        city: 城市名称
21
    """
22
    return f'{city}当天晴朗'
23

24
# 直接传工具名
25
model_with_tools = model.bind_tools([get_weather1, get_weather2], tool_choice="get_weather2")
26

27
messages = [HumanMessage("杭州今天天气如何？")]
28
response = model_with_tools.invoke(messages)
29
response.pretty_print()

1
# ❌ 不好：一个工具做太多事
2
@tool
3
def do_everything(action: str, data: str) -> str:
4
    """做各种事情"""
5
    if action == "weather": ...
6
    elif action == "calculate": ...
7
    elif action == "search": ...
8

9
# ✅ 好：每个工具做一件事
10
@tool
11
def get_weather(city: str) -> str:
12
    """获取天气"""
13
    ...
14

15
@tool
16
def calculator(operation: str, a: float, b: float) -> str:
17
    """计算"""
18
    ...

1
@tool
2
def divide(a: float, b: float) -> str:
3
    """
4
    除法计算
5

6
    Args:
7
        a: 被除数
8
        b: 除数
9
    """
10
    try:
11
        if b == 0:
12
            return "错误：除数不能为零"
13
        result = a / b
14
        return f"{a} / {b} = {result}"
15
    except Exception as e:
16
        return f"计算错误：{e}"