1
tools_by_name = {tool.name: tool for tool in tools}
2
def tool_node(state: dict):
3
    result = []
4
    for tool_call in state["messages"][-1].tool_calls:
5
        tool = tools_by_name[tool_call["name"]]
6
        observation = tool.invoke(tool_call["args"])
7
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
8
    return {"messages": result}

1
import requests
2
import json
3

4
def fetch_real_time_info(query):
5
  """Get real-time Internet information"""
6

7
  url = "https://google.serper.dev/search"
8

9
  payload = json.dumps({
10
    "q": query,
11
    "num": 1,
12
  })
13

14
  headers = {
15
    'X-API-KEY': 'xxx', # 填写自己的serper-key, 可以进入 url 的网址中注册账号获取
16
    'Content-Type': 'application/json'
17
  }
18

19
  response = requests.post(url, headers=headers, data=payload, timeout=10)
20

21
  data = json.loads(response.text)
22

23
  if 'organic' in data:
24
      return json.dumps(data['organic'],  ensure_ascii=False)  # 返回'organic'部分的JSON字符串
25
  else:
26
      return json.dumps({"error": "No organic results found"},  ensure_ascii=False)  # 如果没有'organic'键，返回错误信息
27

28
print(fetch_real_time_info("DeepSeek的最新新闻"))

1
[{"title": "更新日志", "link": "https://api-docs.deepseek.com/zh-cn/updates", "snippet": "新闻. DeepSeek-V3.2-Exp 发布2025/09/29 · DeepSeek V3.1 更新2025/09/22 · DeepSeek V3.1 发布2025/08/21 · DeepSeek-R1-0528 发布2025/05/28 · DeepSeek-V3-0324 发布 ...", "position": 1}]

1
from langchain_core.tools import tool
2
import requests
3
import json
4

5
@tool
6
def fetch_real_time_info(query):
7
  """Get real-time Internet information"""
8

9
  url = "https://google.serper.dev/search"
10

11
  payload = json.dumps({
12
    "q": query,
13
    "num": 1,
14
  })
15

16
  headers = {
17
    'X-API-KEY': 'xxx', # 使用自己的api-key
18
    'Content-Type': 'application/json'
19
  }
20

21
  response = requests.post(url, headers=headers, data=payload, timeout=10)
22

23
  data = json.loads(response.text)
24

25
  if 'organic' in data:
26
      return json.dumps(data['organic'],  ensure_ascii=False)  # 返回'organic'部分的JSON字符串
27
  else:
28
      return json.dumps({"error": "No organic results found"},  ensure_ascii=False)  # 如果没有'organic'键，返回错误信息
29

30
print(f"""name: {fetch_real_time_info.name}
31
description: {fetch_real_time_info.description}
32
arguments: {fetch_real_time_info.args}""")

1
name: fetch_real_time_info
2
description: Get real-time Internet information
3
arguments: {'query': {'title': 'Query'}}

1
from langgraph.prebuilt import ToolNode
2

3
tools = [fetch_real_time_info]
4
tool_node = ToolNode(tools)
5
print(tool_node)

1
tools(tags=None, recurse=True, explode_args=False, func_accepts_config=True, func_accepts={'store': ('__pregel_store', None)}, tools_by_name={'fetch_real_time_info': StructuredTool(name='fetch_real_time_info', description='Get real-time Internet information', args_schema=<class 'langchain_core.utils.pydantic.fetch_real_time_info'>, func=<function fetch_real_time_info at 0x00000238629AD1C0>)}, tool_to_state_args={'fetch_real_time_info': {}}, tool_to_store_arg={'fetch_real_time_info': None}, handle_tool_errors=True, messages_key='messages')

1
from langchain_openai import ChatOpenAI
2
import os
3

4
os.environ["OPENAI_API_KEY"] = "sk-xxx" # 自己的api-key
5
llm = ChatOpenAI(
6
  base_url="https://api.deepseek.com/v1",
7
  model="deepseek-chat",
8
  temperature=0
9
)
10

11
model_with_tools = llm.bind_tools(tools)
12
print(model_with_tools)

1
bound=ChatOpenAI(client=<openai.resources.chat.completions.completions.Completions object at 0x000001D4A259DAF0>, async_client=<openai.resources.chat.completions.completions.AsyncCompletions object at 0x000001D4A2502E10>, root_client=<openai.OpenAI object at 0x000001D4A10CA570>, root_async_client=<openai.AsyncOpenAI object at 0x000001D4A259DF70>, model_name='deepseek-chat', temperature=0.0, model_kwargs={}, openai_api_key=SecretStr('**********'), openai_api_base='https://api.deepseek.com/v1') kwargs={'tools': [{'type': 'function', 'function': {'name': 'fetch_real_time_info', 'description': 'Get real-time Internet information', 'parameters': {'properties': {'query': {}}, 'required': ['query'], 'type': 'object'}}}, {'type': 'function', 'function': {'name': 'get_weather', 'description': 'Get the weather of a location', 'parameters': {'properties': {'location': {}}, 'required': ['location'], 'type': 'object'}}}]} config={} config_factories=[]

1
# 大模型会自动判断需要调用哪个工具
2
response = model_with_tools.invoke("DeepSeek的最新新闻")
3
print(response.tool_calls) # 输出工具调用信息
4

5
# 可以直接传递给ToolNode执行
6
print(tool_node.invoke({"messages": [response]}))

1
[{'name': 'fetch_real_time_info', 'args': {'query': 'DeepSeek最新新闻'}, 'id': 'call_00_LmIQTb8iSy0BHM7iuQPoZV9N', 'type': 'tool_call'}]
2
{'messages': [ToolMessage(content='[{"title": "DeepSeek解决AI有无问题，灵光攻克AI好用难题", "link": "https://t.cj.sina.cn/articles/view/6328613060/m17936f8c403301lueq?from=tech&vt=4", "snippet": "如果说DeepSeek解决了AI“有没有”的问题，那么灵光则解决了AI“好不好用”的难题. 打开App看更多精彩内容. 分享文章到.", "date": "19 hours ago", "position": 1}]', name='fetch_real_time_info', tool_call_id='call_00_LmIQTb8iSy0BHM7iuQPoZV9N')]}

1
import json
2
from tokenize import String
3
from langchain_core.tools import tool
4
from pydantic import Field, BaseModel
5
import requests
6
from langgraph.prebuilt import ToolNode
7
from typing import Optional
8

9
from sqlalchemy import Column, Integer, String, create_engine, text
10
from sqlalchemy.orm import declarative_base, sessionmaker
11

12
# 定义搜索查询模型
13
class SearchQuery(BaseModel):
14
  query: str = Field(description="Questions for networking queries")
15

16
@tool(args_schema=SearchQuery)
17
def fetch_real_time_info(query):
18
  """Get real-time Internet information"""
19
  url = "https://google.serper.dev/search"
20
  payload = json.dumps({"q": query, "num": 1})
21
  headers = {
22
    'X-API-KEY': 'xxx', # 你的api-key
23
    'Content-Type': 'application/json'
24
  }
25
  response = requests.post(url, headers=headers, data=payload)
26
  data = json.loads(response.text)
27
  if 'organic' in data:
28
    return json.dumps(data['organic'], ensure_ascii=False)
29
  else:
30
    return json.dumps({"error": "No organic results found"}, ensure_ascii=False)
31

32
class WeatherLoc(BaseModel):
33
  location: str = Field(description="The location name of the city")
34

35
# 定义获取天气工具
36
@tool(args_schema=WeatherLoc)
37
def get_weather(location):
38
  """Get the weather of a location"""
39
  if location.lower() in ["beijing"]:
40
    return "北京的温度是16度，天气晴朗。"
41
  elif location.lower() in ["shanghai"]:
42
    return "上海的温度是20度，部分多云。"
43
  else:
44
    return "不好意思，并未查询到具体的天气信息。"
45

46
# 创建基类
47
Base = declarative_base()
48

49
# 定义 UserInfo 模型
50
class User(Base):
51
  __tablename__ = 'users'
52
  id = Column(Integer, primary_key=True)
53
  name = Column(String(50))
54
  age = Column(Integer)
55
  email = Column(String(100))
56
  phone = Column(String(15))
57

58
class UserInfo(BaseModel):
59
  """Extracted user information, such as name, age, email, and phone number, if relevant."""
60
  name: str = Field(description="The name of the user")
61
  age: Optional[int] = Field(description="The age of the user")
62
  email: str = Field(description="The email address of the user")
63
  phone: Optional[str] = Field(description="The phone number of the user")
64

65
# MySQL 连接信息（不包含数据库名，用于创建数据库）
66
MYSQL_USER = 'root'
67
MYSQL_PASSWORD = '123456'
68
MYSQL_HOST = 'localhost'
69
MYSQL_PORT = '3306'
70
DATABASE_NAME = 'langgraph_agent'
71

72
# 先连接到 MySQL 服务器（不指定数据库），用于创建数据库
73
server_uri = f'mysql+pymysql://{MYSQL_USER}:{MYSQL_PASSWORD}@{MYSQL_HOST}:{MYSQL_PORT}?charset=utf8mb4'
74
server_engine = create_engine(server_uri, echo=False)
75

76
# 创建数据库（如果不存在）
77
with server_engine.connect() as conn:
78
    conn.execute(text(f"CREATE DATABASE IF NOT EXISTS `{DATABASE_NAME}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"))
79
    conn.commit()
80

81
# 连接到指定的数据库
82
DATABASE_URI = f'mysql+pymysql://{MYSQL_USER}:{MYSQL_PASSWORD}@{MYSQL_HOST}:{MYSQL_PORT}/{DATABASE_NAME}?charset=utf8mb4'
83
engine = create_engine(DATABASE_URI, echo=True)
84

85
# 如果表不存在，则创建表
86
Base.metadata.create_all(engine)
87

88
# 创建会话
89
Session = sessionmaker(bind=engine)
90
session = Session()
91

92
# 定义插入数据库工具
93
@tool(args_schema=UserInfo)
94
def insert_db(name, age, email, phone):
95
  """Insert user information into the database, The required parameters are name, age, email, phone"""
96
  session = Session()
97
  try:
98
    user = User(name=name, age=age, email=email, phone=phone)
99
    session.add(user)
100
    session.commit()
101
    return "用户信息已成功插入数据库。"
102
  except Exception as e:
103
    session.rollback()
104
    return f"用户信息插入失败: {e}"
105
  finally:
106
    session.close()

1
tools = [fetch_real_time_info, get_weather, insert_db]
2
tool_node = ToolNode(tools)
3
print(tool_node)

1
tools(tags=None, recurse=True, explode_args=False, func_accepts_config=True, func_accepts={'store': ('__pregel_store', None)}, tools_by_name={'fetch_real_time_info': StructuredTool(name='fetch_real_time_info', description='Get real-time Internet information', args_schema=<class '__main__.SearchQuery'>, func=<function fetch_real_time_info at 0x00000239B8965300>), 'get_weather': StructuredTool(name='get_weather', description='Get the weather of a location', args_schema=<class '__main__.WeatherLoc'>, func=<function get_weather at 0x00000239B89B9580>), 'insert_db': StructuredTool(name='insert_db', description='Insert user information into the database, The required parameters are name, age, email, phone', args_schema=<class '__main__.UserInfo'>, func=<function insert_db at 0x00000239B8BCDC60>)}, tool_to_state_args={'fetch_real_time_info': {}, 'get_weather': {}, 'insert_db': {}}, tool_to_store_arg={'fetch_real_time_info': None, 'get_weather': None, 'insert_db': None}, handle_tool_errors=True, messages_key='messages')

1
from langchain_openai import ChatOpenAI
2
import os
3

4
os.environ["OPENAI_API_KEY"] = "sk-8317b86e75794ba2a470cfd45c71513d"
5

6
llm = ChatOpenAI(
7
  base_url="https://api.deepseek.com/v1",
8
  model="deepseek-chat",
9
  temperature=0
10
)
11

12
model_with_tools = llm.bind_tools(tools)

1
from typing import Union
2

3
# 定义正常响应模型，可以是用户信息或一般响应
4
class ConversationalResponse(BaseModel):
5
  """Respond to the user's query in a conversational manner. Be kind and helpful."""
6
  response: str = Field(description="A conversational response to the user's query")
7

8
# 定义最终响应模型，可以是用户信息或一般响应
9
class FinalResponse(BaseModel):
10
  final_output: Union[ConversationalResponse, SearchQuery, WeatherLoc, UserInfo]

1
def chat_with_model(state):
2
  """generate structured output"""
3
  print(state)
4
  print("-----------------")
5
  messages = state['messages']
6
  structured_llm = llm.with_structured_output(FinalResponse, method="function_calling")
7
  response = structured_llm.invoke(messages)
8
  return {"messages": [response]}
9

10
def final_answer(state):
11
  """generate natural language responses"""
12
  print(state)
13
  print("-----------------")
14
  messages = state['messages'][-1]
15
  response = messages.final_output.response
16
  return {"messages": [response]}
17

18
def execute_function(state):
19
  """execute tool calling"""
20
  print(state)
21
  print("-----------------")
22
  final_output = state['messages'][-1].final_output
23

24
  # 将结构化输出转换为 AIMessage，包含工具调用
25
  from langchain_core.messages import AIMessage
26

27
  # 根据不同的输出类型构建工具调用
28
  tool_calls = []
29
  if isinstance(final_output, SearchQuery):
30
    tool_calls.append({
31
      "name": "fetch_real_time_info",
32
      "args": {"query": final_output.query},
33
      "id": "tool_call_search",
34
      "type": "tool_call",
35
    })
36
  elif isinstance(final_output, WeatherLoc):
37
    tool_calls.append({
38
      "name": "get_weather",
39
      "args": {"location": final_output.location},
40
      "id": "tool_call_weather",
41
      "type": "tool_call",
42
    })
43
  elif isinstance(final_output, UserInfo):
44
    tool_calls.append({
45
      "name": "insert_db",
46
      "args": {
47
        "name": final_output.name,
48
        "age": final_output.age,
49
        "email": final_output.email,
50
        "phone": final_output.phone
51
      },
52
      "id": "tool_call_db",
53
      "type": "tool_call",
54
    })
55

56
  # 创建包含工具调用的 AIMessage
57
  ai_message = AIMessage(content="", tool_calls=tool_calls)
58

59
  # 调用工具节点
60
  tool_response = tool_node.invoke({"messages": [ai_message]})
61

62
  # 提取工具执行结果
63
  if tool_response.get("messages") and len(tool_response["messages"]) > 0:
64
    # ToolMessage 的内容就是工具的执行结果
65
    result = tool_response["messages"][0].content
66
  else:
67
    result = "工具执行失败：未返回结果"
68

69
  print("工具执行结果:", result)
70
  return {"messages": [result]}

1
from typing import TypedDict, Annotated
2
import operator
3
from langchain_core.messages import AnyMessage
4

5
class AgentState(TypedDict):
6
  messages: Annotated[list[AnyMessage], operator.add]

1
def generate_branch(state: AgentState):
2
  result = state['messages'][-1]
3
  output = result.final_output
4
  if isinstance(output, ConversationalResponse):
5
    return False # 不需要工具调用
6
  else:
7
    return True # 需要工具调用

1
# 构建图结构
2
from langgraph.graph import StateGraph
3

4
graph = StateGraph(AgentState)
5

6
graph.add_node("chat_with_model", chat_with_model)
7
graph.add_node("final_answer", final_answer)
8
graph.add_node("execute_function", execute_function)
9

10
graph.set_entry_point("chat_with_model")
11

12
graph.add_conditional_edges(
13
  "chat_with_model",
14
  generate_branch,
15
  {True: "execute_function", False: "final_answer"}
16
)
17

18
graph.set_finish_point("final_answer")
19
graph.set_finish_point("execute_function")
20

21
graph = graph.compile()

1
from langchain_core.messages import HumanMessage
2

3
# 测试1: 普通对话（不需要工具）
4
query = "你好，请你介绍一下你自己"
5
input_message = {"messages": [HumanMessage(content=query)]}
6
result = graph.invoke(input_message)
7
print("-----------------")
8
print(result)
9
print("-----------------")
10
print(result["messages"][-1])
11

12
# 测试2: 网络搜索（需要工具）
13
query = "帮我查一下DeepSeek的最新新闻"
14
input_message = {"messages": [HumanMessage(content=query)]}
15
result = graph.invoke(input_message)
16
print("-----------------")
17
print(result)
18
print("-----------------")
19
print(result["messages"][-1])
20

21
# 测试3: 天气查询（需要工具）
22
query = "Beijing的天气怎么样？"
23
input_message = {"messages": [HumanMessage(content=query)]}
24
result = graph.invoke(input_message)
25
print("-----------------")
26
print(result)
27
print("-----------------")
28
print(result["messages"][-1])
29

30
# 测试4: 数据库插入（需要工具）
31
query = "我是Hygen，今年18，电话号是01000721，邮箱是ciallo@qq.com"
32
input_message = {"messages": [HumanMessage(content=query)]}
33
result = graph.invoke(input_message)
34
print("-----------------")
35
print(result)
36
print("-----------------")
37
print(result["messages"][-1])

1
{'messages': [HumanMessage(content='你好，请你介绍一下你自己', additional_kwargs={}, response_metadata={})]}
2
-----------------
3
{'messages': [HumanMessage(content='你好，请你介绍一下你自己', additional_kwargs={}, response_metadata={}), FinalResponse(final_output=ConversationalResponse(response='你好！我是一个AI助手，很高兴认识你！我可以帮助你回答各种问题、提供信息、协助解决问题，或者只是和你聊天。我具备多种能力，包
4
括搜索信息、处理文本、分析数据等。\n\n我的特点是：\n- 能够理解和回应中文\n- 可以处理各种类型的问题\n- 提供准确和有用的信息\n- 保持友好和专业的交流方式\n\n如果你有任何问题或需要帮助，请随时告诉我！你想了解什么具体内容，或者有什么我可以为你做的吗？'))]}
5
-----------------
6
-----------------
7
{'messages': [HumanMessage(content='你好，请你介绍一下你自己', additional_kwargs={}, response_metadata={}), FinalResponse(final_output=ConversationalResponse(response='你好！我是一个AI助手，很高兴认识你！我可以帮助你回答各种问题、提供信息、协助解决问题，或者只是和你聊天。我具备多种能力，包 括搜索信息、处理文本、分析数据等。\n\n我的特点是：\n- 能够理解和回应中文\n- 可以处理各种类型的问题\n- 提供准确和有用的信息\n- 保持友好和专业的交流方式\n\n如果你有任何问题或需要帮助，请随时告诉我！你想了解什么具体内容，或者有什么我可以为你做的吗？')), '你好！我是一个AI助手，很高兴认识你！我 可以帮助你回答各种问题、提供信息、协助解决问题，或者只是和你聊天。我具备多种能力，包括搜索信息、处理文本、分析数据等。\n\n我的特点是：\n- 能够理解和回应中文\n- 可以处理各种类型的问题\n- 提供准确和有用的信息\n- 保持友好和专业的交流方式\n\n如果你有任何问题或需要帮助，请随时告诉我！你想了解什 么具体内容，或者有什么我可以为你做的吗？']}
8
-----------------
9
你好！我是一个AI助手，很高兴认识你！我可以帮助你回答各种问题、提供信息、协助解决问题，或者只是和你聊天。我具备多种能力，包括搜索信息、处理文本、分析数据等。
10

11
我的特点是：
12
- 能够理解和回应中文
13
- 可以处理各种类型的问题
14
- 提供准确和有用的信息
15
- 保持友好和专业的交流方式
16

17
如果你有任何问题或需要帮助，请随时告诉我！你想了解什么具体内容，或者有什么我可以为你做的吗？
18
{'messages': [HumanMessage(content='帮我查一下DeepSeek的最新新闻', additional_kwargs={}, response_metadata={})]}
19
-----------------
20
{'messages': [HumanMessage(content='帮我查一下DeepSeek的最新新闻', additional_kwargs={}, response_metadata={}), FinalResponse(final_output=SearchQuery(query='DeepSeek最新新闻'))]}
21
-----------------
22
工具执行结果: [{"title": "DeepSeek等中製AI模型操弄資訊、洩個資國安局", "link": "https://www.cna.com.tw/news/aipl/202511160037.aspx", "snippet": "國安局今天表示，依法抽驗包括DeepSeek等5款中製「生成式AI語言模型」後，所生成內容出現中共官宣、歷史偏差、資訊操弄等嚴重偏頗與不實資訊， ...", "date": "3 days ago", "position": 1}]
23
-----------------
24
{'messages': [HumanMessage(content='帮我查一下DeepSeek的最新新闻', additional_kwargs={}, response_metadata={}), FinalResponse(final_output=SearchQuery(query='DeepSeek最新新闻')), '[{"title": "DeepSeek等中製AI模型操弄資訊、洩個資國安局", "link": "https://www.cna.com.tw/news/aipl/202511160037.aspx", "snippet": "國安局今天表示，依法抽驗包括DeepSeek等5款中製「生成式AI語言模型」後，所生成內容出現中共官宣、歷史偏差、資訊操弄等嚴重偏頗與不實資訊， ...", "date": "3 days ago", "position": 1}]']}
25
-----------------
26
[{"title": "DeepSeek等中製AI模型操弄資訊、洩個資國安局", "link": "https://www.cna.com.tw/news/aipl/202511160037.aspx", "snippet": "國安局今天表示，依法抽驗包括DeepSeek等5款中製「生成式AI語言模型」後，所生成內容出現中共官宣、歷史偏差、資訊操弄等嚴重偏頗與不實資訊， ...", "date": "3 days ago", "position": 1}]
27
{'messages': [HumanMessage(content='Beijing的天气怎么样？', additional_kwargs={}, response_metadata={})]}
28
-----------------
29
{'messages': [HumanMessage(content='Beijing的天气怎么样？', additional_kwargs={}, response_metadata={}), FinalResponse(final_output=WeatherLoc(location='Beijing'))]}
30
-----------------
31
工具执行结果: 北京的温度是16度，天气晴朗。
32
-----------------
33
{'messages': [HumanMessage(content='Beijing的天气怎么样？', additional_kwargs={}, response_metadata={}), FinalResponse(final_output=WeatherLoc(location='Beijing')), '北京的温度是16度，天气晴朗。']}
34
-----------------
35
北京的温度是16度，天气晴朗。
36
{'messages': [HumanMessage(content='我是Hygen，今年18，电话号是01000721，邮箱是ciallo@qq.com', additional_kwargs={}, response_metadata={})]}
37
-----------------
38
{'messages': [HumanMessage(content='我是Hygen，今年18，电话号是01000721，邮箱是ciallo@qq.com', additional_kwargs={}, response_metadata={}), FinalResponse(final_output=UserInfo(name='Hygen', age=18, email='ciallo@qq.com', phone='01000721'))]}
39
-----------------
40
2025-11-19 13:04:06,678 INFO sqlalchemy.engine.Engine BEGIN (implicit)
41
2025-11-19 13:04:06,681 INFO sqlalchemy.engine.Engine INSERT INTO users (name, age, email, phone) VALUES (%(name)s, %(age)s, %(email)s, %(phone)s)
42
2025-11-19 13:04:06,681 INFO sqlalchemy.engine.Engine [generated in 0.00024s] {'name': 'Hygen', 'age': 18, 'email': 'ciallo@qq.com', 'phone': '01000721'}
43
2025-11-19 13:04:06,696 INFO sqlalchemy.engine.Engine COMMIT
44
工具执行结果: 用户信息已成功插入数据库。
45
-----------------
46
{'messages': [HumanMessage(content='我是Hygen，今年18，电话号是01000721，邮箱是ciallo@qq.com', additional_kwargs={}, response_metadata={}), FinalResponse(final_output=UserInfo(name='Hygen', age=18, email='ciallo@qq.com', phone='01000721')), '用户信息已成功插入数据库。']}
47
-----------------
48
用户信息已成功插入数据库。

1
from langchain_core.messages import ToolMessage
2
from langchain_openai import ChatOpenAI
3
import os
4

5
os.environ["OPENAI_API_KEY"] = "sk-8317b86e75794ba2a470cfd45c71513d"
6

7
llm = ChatOpenAI(
8
  base_url="https://api.deepseek.com/v1",
9
  model="deepseek-chat",
10
  temperature=0
11
)
12

13
def chat_with_model(state):
14
  """generate structured output"""
15
  messages = state['messages']
16
  response = llm.invoke(messages)
17
  return {"messages": [response]}
18

19
def execute_function(state):
20
  """execute tool calling"""
21
  tool_calls = state['messages'][-1].tool_calls
22
  results = []
23
  tools = [fetch_real_time_info, get_weather, insert_db]
24
  tools = {t.name: t for t in tools}
25
  for t in tool_calls:
26
    if not t['name'] in tools:
27
      result = "bad tool name, retry"
28
    else:
29
      result = tools[t['name']].invoke(t['args'])
30
    results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
31
  return {"messages": results}

1
def exists_function_calling(state: AgentState):
2
  result = state['messages'][-1]
3
  return len(result.tool_calls) > 0

1
SYSTEM_PROMPT = """
2
Please summarize the information obtained so far and generate a professional response. Note, please reply in Chinese.
3
"""
4

5
from langchain_core.messages import SystemMessage, HumanMessage
6
def natural_response(state):
7
  """generate final language responses"""
8
  messages = state['messages'][-1]
9
  messages = [SystemMessage(content=SYSTEM_PROMPT)] + [HumanMessage(content=messages.content)]
10
  response = llm.invoke(messages)
11
  return {"messages": [response]}

1
from typing import TypedDict, Annotated
2
import operator
3
from langchain_core.messages import AnyMessage
4
from langgraph.graph import StateGraph
5

6
class AgentState(TypedDict):
7
  messages: Annotated[list[AnyMessage], operator.add]
8

9
graph = StateGraph(AgentState)
10

11
graph.add_node("chat_with_model", chat_with_model)
12
graph.add_node("execute_function", execute_function)
13
graph.add_node("final_answer", final_answer)
14
graph.add_node("natural_response", natural_response)
15

16
graph.set_entry_point("chat_with_model")
17

18
graph.add_conditional_edges(
19
  "chat_with_model",
20
  exists_function_calling,
21
  {True: "execute_function", False: "final_answer"}
22
)
23

24
graph.add_edge("execute_function", "natural_response")
25
graph.add_edge("final_answer", "natural_response")
26

27
graph.set_finish_point("natural_response")
28

29
graph = graph.compile()

1
tools = [insert_db, fetch_real_time_info, get_weather]
2
llm = llm.bind_tools(tools)

1
# 测试1: 普通对话（不需要工具）
2
query = "你好，请你介绍一下你自己"
3
input_message = {"messages": [HumanMessage(content=query)]}
4
result = graph.invoke(input_message)
5
print("-----------------")
6
print(result)
7

8
# 测试2: 网络搜索（需要工具）
9
query = "帮我查一下DeepSeek的最新新闻"
10
input_message = {"messages": [HumanMessage(content=query)]}
11
result = graph.invoke(input_message)
12
print("-----------------")
13
print(result)
14

15
# 测试3: 天气查询（需要工具）
16
query = "Beijing的天气怎么样？"
17
input_message = {"messages": [HumanMessage(content=query)]}
18
result = graph.invoke(input_message)
19
print("-----------------")
20
print(result)
21

22
# 测试4: 数据库插入（需要工具）
23
query = "我是Hygen，今年28，电话号是01000721，邮箱是ciallo@qq.com"
24
input_message = {"messages": [HumanMessage(content=query)]}
25
result = graph.invoke(input_message)
26
print("-----------------")
27
print(result)

1
-----------------
2
{'messages': [HumanMessage(content='你好，请你介绍一下你自己', additional_kwargs={}, response_metadata={}), AIMessage(content='你好！我是一个AI助手，很高兴认识你！我可以帮助你处理各种任务，包括：\n\n- 获取实时网络信息\n- 查询天气情况\n- 管理用户信息（存储到数据库中）\n\n我可以通过工具来获取最新的网络信息、查询特定城市的天气，以及将用户信息（姓名、年龄、邮箱、电话）保存到数据库中。\n\n请告诉我你需要什么帮助，我会尽力为你提供支持！', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 84, 'prompt_tokens': 368, 'total_tokens': 452, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 320}, 'prompt_cache_hit_tokens': 320, 'prompt_cache_miss_tokens': 48}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '94aaa2f7-b65e-458c-b369-a70d841f105e', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--ee81ac9d-e4f0-492c-95db-1928d9c54202-0', usage_metadata={'input_tokens': 368, 'output_tokens': 84, 'total_tokens': 452, 'input_token_details': {'cache_read': 320}, 'output_token_details': {}}), AIMessage(content='你好！我是一个AI助手，很高兴认识你！我可以帮助你处理各种任务，包括：\n\n- 获取实时网络信息\n- 查询天气情况\n- 管理用户信息（存储到数据库中）\n\n我可以通过工具来获取最新的网络信息、查询特定城市的天气，以及将用户信息（姓名、年龄、邮箱、电话）保存到数据库中。\n\n请告诉我你需要什么帮助 ，我会尽力为你提供支持！', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 84, 'prompt_tokens': 368, 'total_tokens': 452, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 320}, 'prompt_cache_hit_tokens': 320, 'prompt_cache_miss_tokens': 48}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '94aaa2f7-b65e-458c-b369-a70d841f105e', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--ee81ac9d-e4f0-492c-95db-1928d9c54202-0', usage_metadata={'input_tokens': 368, 'output_tokens': 84, 'total_tokens': 452, 'input_token_details': {'cache_read': 320}, 'output_token_details': {}}), AIMessage(content='你好！很高兴认识你！感谢你详细介绍你的功能和服务范围。\n\n根据你的描述，你确实具备以下强大的能力：\n\n**主要功能：**\n1. **实时网络信息查询** - 可以获取最新的网络资讯和信息\n2. **天气查询** - 能够查询特定城市的天气情况\n3. **用户信息管理** - 可以将用户信息（姓名、年龄、邮箱、电话）存储到数据库中\n\n**使用方式：**\n- 如果你需要查询任何实时信息，我可以帮你搜索网络\n- 如果你想知道某个城市的天气，告诉 我城市名即可\n- 如果你需要保存用户信息，请提供姓名、年龄、邮箱和电话\n\n我现在已经准备好为你提供帮助了！请告诉我你需要什么服务，或者有什么具体的问题需要解决？我会根据你的需求选择最合适的工具来协助你。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 170, 'prompt_tokens': 468, 'total_tokens': 638, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 468}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '8f3dbeff-b403-4aef-a685-1bc425d6c62a', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--092e834b-2c6c-4d8a-a5ea-b4fb2df4f761-0', usage_metadata={'input_tokens': 468, 'output_tokens': 170, 'total_tokens': 638, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})]}
3
-----------------
4
{'messages': [HumanMessage(content='帮我查一下DeepSeek的最新新闻', additional_kwargs={}, response_metadata={}), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_00_EM6RKCZJbx09rIEDrVA0Re0L', 'function': {'arguments': '{"query": "DeepSeek最新新闻"}', 'name': 'fetch_real_time_info'}, 'type': 'function', 'index': 0}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 19, 'prompt_tokens': 371, 'total_tokens': 390, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 320}, 'prompt_cache_hit_tokens': 320, 'prompt_cache_miss_tokens': 51}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': 'a1826f70-cfa1-4fab-8412-09bc3f901a06', 'service_tier': None, 'finish_reason': 'tool_calls', 'logprobs': None}, id='run--a95bf745-8601-4fe5-bc77-4e2014e508cf-0', tool_calls=[{'name': 'fetch_real_time_info', 'args': {'query': 'DeepSeek最新新闻'}, 'id': 'call_00_EM6RKCZJbx09rIEDrVA0Re0L', 'type': 'tool_call'}], usage_metadata={'input_tokens': 371, 'output_tokens': 19, 'total_tokens': 390, 'input_token_details': {'cache_read': 320}, 'output_token_details': {}}), ToolMessage(content='[{"title": "DeepSeek | 世界新聞網", "link": "https://www.worldjournal.com/search/tagging/8877/DeepSeek", "snippet": "FT：DeepSeek新模型延後問世問題出在華為晶片還是得靠輝達 · 2025-08-14 03:00 ; OpenAI最新模型GPT-5發布傳DeepSeek將推R2 · 2025-08-14 02:04 ; 受惠DeepSeek-R2將發表中國AI ...", "position": 1}]', name='fetch_real_time_info', tool_call_id='call_00_EM6RKCZJbx09rIEDrVA0Re0L'), AIMessage(content='根据您提供的信息，我看到了关于DeepSeek的一些最新新闻摘要。这些信息主要来自世界新聞網，涉及以下几个要点：\n\n1. **DeepSeek新模型延后问世** - 有报道讨论其延迟原因是否与华为芯片相关还是需要依赖辉达（NVIDIA）\n\n2. **OpenAI GPT-5发布** - 同时有传闻DeepSeek将推出R2模型\n\n3. **中国AI受益** - 提到受惠于DeepSeek-R2即将发表\n\n不过，这些信息相对有限，主要是新闻标题和简短摘要。如果您需要更详细、更全面的DeepSeek相关信息，我可以 帮您搜索最新的实时信息来获取更完整的报道内容。\n\n您希望我为您搜索更多关于DeepSeek的详细资讯吗？', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 162, 'prompt_tokens': 509, 'total_tokens': 671, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 320}, 'prompt_cache_hit_tokens': 320, 'prompt_cache_miss_tokens': 189}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '0f5dd1f4-4b97-41a8-bd61-81943a5da6bb', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--19b47710-a7fb-4d37-9d7d-07b58c48f4e5-0', usage_metadata={'input_tokens': 509, 'output_tokens': 162, 'total_tokens': 671, 'input_token_details': {'cache_read': 320}, 'output_token_details': {}})]}
5
-----------------
6
{'messages': [HumanMessage(content='Beijing的天气怎么样？', additional_kwargs={}, response_metadata={}), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_00_QxjiQuICuIOKzOOgQaJCAcGo', 'function': {'arguments': '{"location": "Beijing"}', 'name': 'get_weather'}, 'type': 'function', 'index': 0}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 368, 'total_tokens': 382, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 320}, 'prompt_cache_hit_tokens': 320, 'prompt_cache_miss_tokens': 48}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '16c53fc1-d392-4626-8268-e6104fab5bae', 'service_tier': None, 'finish_reason': 'tool_calls', 'logprobs': None}, id='run--76f8f809-a413-48fb-8a30-9401700c3fd3-0', tool_calls=[{'name': 'get_weather', 'args': {'location': 'Beijing'}, 'id': 'call_00_QxjiQuICuIOKzOOgQaJCAcGo', 'type': 'tool_call'}], usage_metadata={'input_tokens': 368, 'output_tokens': 14, 'total_tokens': 382, 'input_token_details': {'cache_read': 320}, 'output_token_details': {}}), ToolMessage(content='北京的温度是16度，天气晴朗。', name='get_weather', tool_call_id='call_00_QxjiQuICuIOKzOOgQaJCAcGo'), AIMessage(content='根据您提供的信息，北京当前的温度是16度，天气晴朗。这是一个非常宜人的天气状况，适合户外活动和出行。\n\n如果您需要获取更详细的天气信息，比如湿度、风速、空气质量指 数等，我可以帮您查询最新的实时天气数据。您是否需要我为您获取更全面的北京天气信息？', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 64, 'prompt_tokens': 393, 'total_tokens': 457, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 320}, 'prompt_cache_hit_tokens': 320, 'prompt_cache_miss_tokens': 73}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': 'bfd56895-dda9-4e7e-9a48-99e105c185e1', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--bc232be5-2b56-44f5-801c-454435c722ab-0', usage_metadata={'input_tokens': 393, 'output_tokens': 64, 'total_tokens': 457, 'input_token_details': {'cache_read': 320}, 'output_token_details': {}})]}
7
2025-11-19 13:27:10,184 INFO sqlalchemy.engine.Engine BEGIN (implicit)
8
2025-11-19 13:27:10,187 INFO sqlalchemy.engine.Engine INSERT INTO users (name, age, email, phone) VALUES (%(name)s, %(age)s, %(email)s, %(phone)s)
9
2025-11-19 13:27:10,189 INFO sqlalchemy.engine.Engine [generated in 0.00027s] {'name': 'Hygen', 'age': 28, 'email': 'ciallo@qq.com', 'phone': '01000721'}
10
2025-11-19 13:27:10,191 INFO sqlalchemy.engine.Engine COMMIT
11
-----------------
12
{'messages': [HumanMessage(content='我是Hygen，今年28，电话号是01000721，邮箱是ciallo@qq.com', additional_kwargs={}, response_metadata={}), AIMessage(content='您好Hygen！我已经收到了您的个人信息：\n- 姓名：Hygen\n- 年龄：28岁\n- 电话：01000721\n- 邮箱：ciallo@qq.com\n\n现在我将把这些信息保 存到数据库中。', additional_kwargs={'tool_calls': [{'id': 'call_00_svoG4YILmEjp7UVtyhHAm7CL', 'function': {'arguments': '{"name": "Hygen", "age": 28, "email": "ciallo@qq.com", "phone": "01000721"}', 'name': 'insert_db'}, 'type': 'function', 'index': 0}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 87, 'prompt_tokens': 384, 'total_tokens': 471, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 320}, 'prompt_cache_hit_tokens': 320, 'prompt_cache_miss_tokens': 64}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '2b5d2240-b1eb-4cff-8ea6-642647c7980c', 'service_tier': None, 'finish_reason': 'tool_calls', 'logprobs': None}, id='run--625e59b0-ec90-455b-94b3-d210c882e61a-0', tool_calls=[{'name': 'insert_db', 'args': {'name': 'Hygen', 'age': 28, 'email': 'ciallo@qq.com', 'phone': '01000721'}, 'id': 'call_00_svoG4YILmEjp7UVtyhHAm7CL', 'type': 'tool_call'}], usage_metadata={'input_tokens': 384, 'output_tokens': 87, 'total_tokens': 471, 'input_token_details': {'cache_read': 320}, 'output_token_details': {}}), ToolMessage(content='用户信息已成功插入数据库。', name='insert_db', tool_call_id='call_00_svoG4YILmEjp7UVtyhHAm7CL'), AIMessage(content='根据您提供的信息，用户数据已成功保存到数据库中。这是一个标准的确认信息，表明：\n\n**操作结果：成功**\n- 用户信息已完整录入数据库系统\n- 数据存储过程顺利完成\n- 系统已确认保存 操作\n\n如果您需要：\n- 查询已保存的用户信息\n- 进行其他数据库操作\n- 获取更多关于数据管理的帮助\n\n请随时告知，我将为您提供相应的支持。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 88, 'prompt_tokens': 391, 'total_tokens': 479, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 320}, 'prompt_cache_hit_tokens': 320, 'prompt_cache_miss_tokens': 71}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '2d284b4f-9eec-49e1-91d2-8e51fb3f94da', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--ceaa01d6-396c-4bf6-8cf4-4644f265ffa6-0', usage_metadata={'input_tokens': 391, 'output_tokens': 88, 'total_tokens': 479, 'input_token_details': {'cache_read': 320}, 'output_token_details': {}})]}