langgraph四大代理架构

一、概念

• 路由代理
• 工具代理
• 自主循环代理
• 多代理

二、路由代理(Router Agent)

I 概念（有了条件边，就意味着有了选择）

在 LangGraph 中，我们可以利用 “条件边” 这一概念来指导或约束大模型在处理特定任务时的逻辑流程。这种机制允许大模型在达到某一环节并满足预设条件时，根据不同的条件输出或数据，选择性地执行不同的逻辑路径。

在 LangGraph 等框架中，条件边赋予了工作流在运行时动态决策的能力。静态边（add_edge）像固定的地铁线路，而条件边则像打车导航，能够根据当前状态（State）实时选择最优路线。

II 有向无环图（DAG）代码演示

只有静态边的智能体框架，通常被称为基于有向无环图（DAG）的框架。
在智能体架构中，如果执行路径的每一步都是提前确定好的，且节点之间仅通过静态边进行单向、无循环的连接，这种结构就被称为 DAG（Directed Acyclic Graph）。它对应的是一种静态编排。如代码所示：

from langgraph.graph import START, StateGraph, END
from langgraph.graph import StateGraph
import os
def node_a(state):
    return {"x": state["x"] + 1}

def node_b(state):
    return {"x": state["x"] - 2}

builder = StateGraph(dict)

builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)

# 构建节点之间的边
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)

graph = builder.compile()

# 渲染流程图可视化
image_data = graph.get_graph().draw_mermaid_png()
# 写入文件
with open("Router_Agent_G.png", "wb") as f:
    f.write(image_data)
print(f"✅ 图片已成功生成: {os.path.abspath("Router_Agent_G.png")}")

为了管理这样复杂的图结构，LangGraph 使用的是一个类似于 if-else 语句的结构组件，称为 Router（路由）。这个组件允许大模型从一组预设选项中选择合适的步骤来进行执行。这个设计思路并不难理解，同时由于 LangGraph 的底层封装，实现起来也非常简单，我们看下面的代码：

如果想选择性地路由到 1 个或多个边，则需要使用 add_conditional_edges 方法。该方法也在 Graph 的基类中进行了定义，如下所示：

Graph 基类源码片段

class Graph:
    def __init__(self) -> None:
        self.nodes: dict[str, NodeSpec] = {}
        self.edges = set[tuple[str, str]]()
        self.branches: defaultdict[str, dict[str, Branch]] = defaultdict(dict)
        self.support_multiple_edges = False
        self.compiled = False

    def add_conditional_edges(
        self,
        source: str,  # 起始节点
        path: Union[  # 这是一个可调用对象，其返回值决定接下来执行的节点。这个函数可以是简单的 Python 函数，或者是任何可以被调用来决定分支路径的对象。
            Callable[..., Union[Hashable, list[Hashable]]],
            Callable[..., Awaitable[Union[Hashable, list[Hashable]]]],
            Runnable[Any, Union[Hashable, list[Hashable]]],
        ],
        path_map: Optional[Union[dict[Hashable, str], list[str]]] = None,  # 路径到节点名称的可选映射。如果省略，path 返回的路径应该是节点名称。
        then: Optional[str] = None,  # 在path选择的节点之后执行的节点的名称。
    ) -> Self:

add_conditional_edges 源码地址：https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.graph.Graph

根据源码的定义我们可以非常明确的分析出其调用过程。这里我们要关注 path 参数，它指的是一个函数调用对象，与普通的节点类似，这个对象接受图的当前 state 并返回一个值，根据返回值的不同，来决定路由到哪个节点。比如想构建一个带有路由的图结构，这里我们定义一个 routing_function 作为路由函数，并添加一个新的节点node_c，代码如下所示：

II 简单的路由代理 代码演示

from langgraph.graph import START, StateGraph, END
from langgraph.graph import StateGraph
from IPython.display import Image, display
import os
from typing import TypedDict, Annotated
from operator import add

class State(TypedDict):
    x: int

def node_a(state):
    return {"x": state["x"] + 1}

def node_b(state):
    return {"x": state["x"] - 2}

def node_c(state):
    return {"x": state["x"] + 1}

def routing_function(state):
    if state["x"] == 10:
        return True
    else:
        return False

builder = StateGraph(State)

builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("node_c", node_c)

builder.set_entry_point("node_a")

# 构建节点之间的边
builder.add_conditional_edges("node_a", routing_function, {
        True: "node_b",
        False: "node_c",
    })
builder.add_edge("node_b", END)
builder.add_edge("node_c", END)

graph = builder.compile()

# 2、渲染流程图可视化
image_data = graph.get_graph().draw_mermaid_png()
# 写入文件
with open("Router_Agent_G.png", "wb") as f:
    f.write(image_data)
print(f"✅ 图片已成功生成: {os.path.abspath("Router_Agent_G.png")}")

II 复杂的路由代理 代码演示（由大模型判断路由【给添加提示词】）

这里的核心是 Router function，它根据输入数据的结构和内容，动态地决定下一步应该执行的节点。例如，对于具体的查询请求，Router 决定需要访问数据库 (Mysql 节点)，而对于简单的问候（如 "Hello"），则直接返回一个响应 (Response 节点)。每个决策路径最终都指向一个结束节点 (End)。所以我们要明确的是，在构建实际的 Agent 时，Router fuction 的定义才是最关键且最重要的。我们需要在这个函数中，基于特定的一些格式或者标识来区分该执行哪一条分支的逻辑。而对于消息的传递，大模型往往是通过结构化输出，引导其在响应的过程中应遵循哪种模式来工作，就类似于工具调用过程。Router 就很好的利用到了这个特性，通过结构化输出的特性来控制接下来的分支路径。

这里我们先来了解一下什么是结构化输出。在 LangGraph 中，实现结构化输出可以通过以下三种有效方式完成：

• 提示工程：指示大模型以特定格式做出回应。
• 输出解析器：采用后处理的方法从大模型的响应中提取结构化数据。
• 工具调用：利用一些内置工具调用功能来生成结构化输出。

1、结构化输出-提示词工程

1.1 给大模型写提示词

# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["QWEN_API_KEY"]
base_url = os.environ["QWEN_BASE_URL"]
model_name = os.environ["QWEN_MODEL_NAME"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model=model_name, api_key=key,base_url=base_url,temperature=0,)

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "Answer the user query. Wrap the output in ``<code>json</code>``",
        ),
        ("human", "{query}"),
    ]
)

chain = prompt | llm

ans = chain.invoke({"query": "我叫奥特曼，今年38岁，邮箱地址是aoteman#qq.com，电话是12111111111"})

print(f"{'*' * 80} \n {ans.content} \n {'*' * 80} ")

"""
打印的结果：

```json
{
  "name": "奥特曼",
  "age": 38,
  "email": "aoteman#qq.com",
  "phone": "12111111111",
  "note": "信息已记录。温馨提示：标准邮箱地址中的“#”通常应为“@”，如需修正请提供正确格式。"
}</code></pre>
<p>"""</p>
<pre><code>
##### 1.2 优化方案，提示+输出解析器（extract_json）

直接通过提示工程让大模型生成特定格式的输出虽然是可行的，但这种方法在复杂的 Agent 构建流程中非常不稳定，
一个优选的优化方法是：通过输出解析器来格式化大模型生成的响应。这种做法可以提高输出的准确性和一致性，这种形式的实现方法如下所示：

```python
# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["QWEN_API_KEY"]
base_url = os.environ["QWEN_BASE_URL"]
model_name = os.environ["QWEN_MODEL_NAME"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model=model_name, api_key=key,base_url=base_url,temperature=0,)

from langchain_core.messages import AIMessage
import json
import re
from typing import List

def extract_json(message: AIMessage) -> List[dict]:
    """Extracts JSON content from a string where JSON is embedded between ``<code>json and </code>`` tags.

    Parameters:
        text (str): The text containing the JSON content.

    Returns:
        list: A list of extracted JSON strings.
    """
    text = message.content
    # 定义正则表达式模式来匹配JSON块
    pattern = r"``<code>json(.*?)</code>``"

    # 在字符串中查找模式的所有非重叠匹配
    matches = re.findall(pattern, text, re.DOTALL)

    # 返回匹配的JSON字符串列表，去掉任何开头或结尾的空格
    try:
        return [json.loads(match.strip()) for match in matches]
    except Exception:
        raise ValueError(f"Failed to parse: {message}")

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "Answer the user query. Wrap the output in ``<code>json</code>``",
        ),
        ("human", "{query}"),
    ]
)

chain = prompt | llm | extract_json

ans = chain.invoke({"query": "我叫奥特曼，今年38岁，邮箱地址是aoteman#qq.com，电话是12111111111"})

print(f"{'*' * 80} \n {ans} \n {'*' * 80} ")

"""
输出的结果：
[{'name': '奥特曼', 'age': 38, 'email': 'aoteman#qq.com', 'phone': '12111111111', 'note': '信息已记录。温馨提示：标准邮箱地址中的“#”通常应为“@”，如需修正请提供正确格式。'}] 
"""

2、langgraph-格式化输出

内置工具方法
从结果可以明显看出，通过定制化的输出解析器得到的结果会更加的符合预期，而在 LangGraph 中更常用的，且效果更好的是，直接使用其内置的工具方法：.with_structured_output()。

这个方法通过接受一个定义了所需输出属性的名称、类型和描述的模式作为输入，进而生成一个类似模型的 Runnable 。不同于常规模型输出字符串或消息，这个 Runnable 输出一个与输入模式相匹配的对象。可以通过几种方式指定这种架构，包括

• TypedDict
• JSON Schema
• Pydantic 类。
  如果采用 TypedDict 或 JSON Schema ，Runnable 将输出一个字典；若使用 Pydantic 类，则输出一个 Pydantic 对象。
  先尝试和实践使用 Pydantic 类做格式化输出，应用的场景是：从文本中提取格式化的数据。

具体内容转入具体文章的链接：https://altairnexus.top/2026/06/17/langgraph-structured-output/

2.1 代码演示

2.1.1 执行语句，将数据插入mysql数据库

掌握到这种程度，我们就可以利用这些判别条件作为 Router Function 来构建决策分支。通过使用 Pydantic 模型来提取结构化数据，能够在大语言模型 (LLM) 调用过程中将非结构化文本转换为结构化数据格式。这种转换使得我们构建的流程图能够根据用户的不同输入，智能判断是应当生成常规响应还是执行数据库操作。代码如下所示：

MySQL 完整安装配置步骤（Ubuntu 系统）

# 更新软件源索引
sudo apt-get update
# 安装 MySQL 服务端
sudo apt-get install mysql-server
# 启动 MySQL 服务
sudo service mysql start

进入 MySQL 客户端，执行账号与库创建 SQL

0. 登录数据库
mysql -u root

1. 创建专用业务账号 gpt
CREATE USER 'gpt'@'localhost' IDENTIFIED BY 'gpt';

2. 给账号授予全库全部权限
GRANT ALL PRIVILEGES ON *.* TO 'gpt'@'localhost';

3. 刷新权限使配置生效
FLUSH PRIVILEGES;

4. 创建项目专用数据库 langgraph
create database langgraph;

Python 依赖安装（对接 MySQL）

pip install sqlalchemy pymysql

MySQL 完整安装配置步骤（windows 系统）

https://altairnexus.top/2026/07/08/mysql_databases/

2.1.2 代码演示

from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData
from sqlalchemy.orm import declarative_base, sessionmaker
from typing import Optional, Literal
from pydantic import BaseModel, Field
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=10)
# 给字段加examples示例，让模型"认得"要提取什么。例子比描述更能锚定模型对字段的理解，尤其对中文输入。
class UserInfo(BaseModel):
    """用户个人信息。当路由判定为 user_info 时填写本字段，缺失字段留空。"""
    name: Optional[str] = Field(default=None, description="用户姓名", examples=["奥特曼", "张三"])
    age: Optional[int] = Field(default=None, description="用户年龄", examples=[38])
    email: Optional[str] = Field(default=None, description="用户邮箱地址", examples=["aoteman@qq.com"])
    phone: Optional[str] = Field(default=None, description="用户电话号码", examples=["12111111111"])

# 定义正常生成模型回复的模型
class ConversationalResponse(BaseModel):
    """对话回复。当路由判定为 chat 时填写本字段。"""
    response: str = Field(description="A conversational response to the user's query")

# 定义最终响应模型：先做显式路由判定，再按路由填写对应字段。
# 用 route 字段做分类，比 Union 交给模型选工具更稳定（兼容接口模型对工具选择的遵循度参差不齐）。
class FinalResponse(BaseModel):
    """先判断用户消息属于哪一类，再填对应字段。
    判定规则：
    - 含姓名/年龄/邮箱/电话/手机号等任意个人信息信号词（我叫/我是/今年X岁/邮箱/电话/手机）→ route="user_info"，并填 user_info。
    - 纯寒暄/提问/闲聊，不含任何个人信息字段（你好/谢谢/解释一下XX/今天天气怎么样）→ route="chat"，并填 chat。"""
    route: Literal["user_info", "chat"] = Field(description="消息类型路由：含个人信息选 user_info，纯对话选 chat")
    user_info: Optional[UserInfo] = Field(default=None, description="route=user_info 时填写")
    chat: Optional[ConversationalResponse] = Field(default=None, description="route=chat 时填写")

# 创建基类
Base = declarative_base()

# 定义 UserInfo 映射模型，对应数据库 users 表
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    age = Column(Integer)
    email = Column(String(100))
    phone = Column(String(15))

# 数据库连接 URI 说明
# gpt: MySQL 用户名
# gpt: MySQL 密码
# localhost: MySQL 服务地址
# langgraph: 目标数据库名
# charset=utf8mb4：支持完整Unicode、emoji字符
# 1、读取配置信息
passwrod = os.environ["MYSQL_PASSWORD"]
acc = os.environ["MYSQL_account"]
DATABASE_URI = f'mysql+pymysql://{acc}:{passwrod}@localhost/langgraph?charset=utf8mb4'
engine = create_engine(DATABASE_URI, echo=True)

# 自动创建数据表（不存在则新建，存在不会覆盖）
Base.metadata.create_all(engine)

# 创建数据库会话工厂
Session = sessionmaker(bind=engine)
session = Session()

from langchain_core.messages import SystemMessage
system = SystemMessage(content=(
    "你是一个消息分类与信息抽取器，先判定 route，再按 route 填对应字段。\n"
    "判定规则（必须严格遵守）：\n"
    "1) 若用户消息中出现姓名/年龄/邮箱/电话/手机号等任意个人信息信号词"
    "（如 我叫/我是/姓名/今年X岁/邮箱/电话/手机），route 必须为 'user_info'，"
    "并在 user_info 中填入抽取到的字段，缺失字段留空；此时 chat 必须为空。\n"
    "2) 若用户消息是纯寒暄/提问/闲聊，不含任何个人信息字段"
    "（如 你好/谢谢/解释一下XX/今天天气怎么样），route 必须为 'chat'，"
    "并在 chat.response 中给出自然语言回复；此时 user_info 必须为空。\n"
    "禁止在含个人信息的消息上选 'chat'，也禁止在纯对话消息上选 'user_info'。"
))

# 接下来我们定义节点函数，其中 structured_output 作为路由节点将用户输入的文本转化成格式化输出，搭配 Router Function 构建分支。
def structured_output(state):
    """generate structured output"""
    print(state)
    print("----------------")
    messages = state['messages']
    structured_llm = llm.with_structured_output(FinalResponse)
    # 给大模型写的提示词很重要，直接决定了路由是否正确，以及后续是否可以把结构性数据存入数据库
    response = structured_llm.invoke([system] + messages)
    return {"messages": [response]}

# 然后分别定义两个分支节点，其中 final_answer 用于直接生成响应，而 insert_db 用于执行数据库插入操作。
def final_answer(state):
    """generate natural language responses"""
    print(state)
    print("----------------")
    result = state['messages'][-1]
    # 兼容两种返回：结构化对象 或 纯字符串
    if isinstance(result, FinalResponse):
        response = result.chat.response if result.chat else ""
    else:
        response = result
    return {"messages": [response]}

def insert_db(state):
    """Insert user information into the database"""
    session = Session()  # 确保为每次操作创建新的会话
    try:
        result = state['messages'][-1]
        info = result.user_info
        # 创建用户实例
        user = User(name=info.name, age=info.age, email=info.email, phone=info.phone)
        # 添加到会话
        session.add(user)
        # 提交事务
        session.commit()
        return {"messages": [f"数据已成功存储至Mysql数据库。"]}
    except Exception as e:
        session.rollback()  # 出错时回滚
        return {"messages": [f"数据存储失败，错误原因：{e}"]}
    finally:
        session.close()  # 关闭会话

# 定义好了所有节点函数后，开始构建图。
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage

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

# 定义 generate_branch 函数作为 Router Function，根据经过 structured_output 节点后产生的 route 字段，选择连接不同的节点，即 final_answer 或 insert_db。
def generate_branch(state: AgentState):
    result = state['messages'][-1]
    # 兜底：结构化失败退化成字符串时退回对话分支
    if not isinstance(result, FinalResponse):
        return False
    return result.route == "user_info"

# 构建图并使用条件边来生成 Router。代码如下:
graph = StateGraph(AgentState)

# 添加三个节点
graph.add_node("structured_output", structured_output)
graph.add_node("final_answer", final_answer)
graph.add_node("insert_db", insert_db)

# 设置图的启动节点
graph.set_entry_point("structured_output")

# 设置条件边
graph.add_conditional_edges(
    "structured_output",
    generate_branch,
    {True: "insert_db", False: "final_answer"}
)

# 设置终止节点
graph.set_finish_point("final_answer")
graph.set_finish_point("insert_db")

# 编译图
graph = graph.compile()
# 3、渲染流程图可视化
image_data = graph.get_graph().draw_mermaid_png()
with open("Router_Mysql_G.png", "wb") as f:
    f.write(image_data)

# 接下来进行测试，首先测试执行插入数据库的条件分支。
query="我叫杰克奥特曼，今年30岁，邮箱地址是jake@qq.com，电话是01010101010101"
input = {"messages": [HumanMessage(content=query)]}
result = graph.invoke(input)

print(f"{"测试执行插入数据库的条件分支："}\n{'*'*80}\n{result}\n{'*'*80}")

# 读取最终输出文本
result = result["messages"][-1]
print(f"{"读取最终输出文本："}\n{'*'*80}\n{result}\n{'*'*80}")
# 返回结果：'数据已成功存储至Mysql数据库。'

# 而如果正常的问答，则会经过 final_answer 直接生成响应。
# query="你好，请你介绍一下你自己"

# 普通问答分支测试1
query = "你好，请你介绍一下你自己"
input_message = {"messages": [HumanMessage(content=query)]}
result = graph.invoke(input_message)
print(f"{"普通问答分支测试1："}\n{'*'*80}\n{result}\n{'*'*80}")
result=result["messages"][-1]
print(f"{"读取最终输出文本："}\n{'*'*80}\n{result}\n{'*'*80}")
# 返回输出：
# '你好！我是一个由人工智能驱动的助手，旨在为您提供信息、回答问题和提供帮助支持。请随时告诉我您需要什么帮助！'

# 普通问答分支测试2
query = "请问什么是机器学习"
input_message = {"messages": [HumanMessage(content=query)]}
result = graph.invoke(input_message)
print(f"{"普通问答分支测试2："}\n{'*'*80}\n{result}\n{'*'*80}")
result = result["messages"][-1]
print(f"{"读取最终输出文本："}\n{'*'*80}\n{result}\n{'*'*80}")

query="我叫孙悟空，今年500岁，邮箱地址是wukong@gmial.com"
input_message = {"messages": [HumanMessage(content=query)]}

result = graph.invoke(input_message)
print(f"{"测试执行插入数据库的条件分支："}\n{'*'*80}\n{result}\n{'*'*80}")

如上示例就是 LangGraph 中 Router 的常用使用形式，通过预定义的分支结构，可以根据用户的输入请求灵活适配不同的场景，在这个过程中，结构化输出对于路由至关重要，因为它们确保系统可以可靠地解释大模型的决定并采取行动。这种 Router Agent（路由代理）的优势就是可以精准的控制程序链路中的每一个细节，但同时也表现出来了这是一种相对有限的控制级别的代理架构，因为大模型通常只能控制单个决策。想象一下上面的场景中，如果我们希望定义的 insert_db 不仅仅只是包含插入数据库，而是有一堆各式各样的工具，比如网络搜索，RAG 等等，应该如何进一步的扩展呢？难道要做对每一个工具在 insert_db 节点下再通过 Router Function 做分支判断吗？虽然可行，但总归并不是高效的做法。
由此，我们接下来进一步给大家介绍 Tool Calling Agent（工具调用代理）来高效的解决这一问题。

三、工具代理(Tool Agent)

I 前言

Tool Calling Agent（工具调用代理）是 LangGraph 支持的第二种 AI Agent 代理架构。这个代理架构是在 Router Agent 的基础上，大模型可以自主选择并使用多种工具来完成某个条件分支中的任务。工具调用大家应该非常熟悉了，当我们希望代理与外部系统交互时，工具就非常有用。外部系统（例如 API）通常需要特定输入模式，而不是自然语言。例如，当我们绑定 API 作为工具时，我们赋予大模型对所需输入模式的感知，大模型就能根据用户的自然语言输入选择调用工具，并返回符合该工具架构的输出。

在 LangGraph 框架中，可以直接使用预构建 ToolNode 进行工具调用，其内部实现原理和我们之前介绍的手动实现的 Function Calling 流程思路基本一致，即：
LangGraph ToolNode 源码: https://langchain-ai.github.io/langgraph/reference/prebuilt/#langgraph.prebuilt.tool_node.ToolNode

1、tool_node源码

tools_by_name = {tool.name: tool for tool in tools}

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

II 正文

1 使用 ToolNode 接入工具

经过 ToolNode 工具后，其返回的是一个 LangChain Runnable 对象，会将图形状态（带有消息列表）作为输入并输出状态更新以及工具调用的结果，通过这种设计去适配 LangGraph 中其他的功能组件。比如我们后续要介绍的 LangGraph 预构建的更高级 AI Agent 架构 – ReAct，两者搭配起来可以开箱即用，同时通过 ToolNode 构建的工具对象也能与任何 StateGraph 一起使用，只要其状态中具有带有适当 Reducer 的 messages 键。由此，对于 ToolNode 的使用，有三个必要的点需要满足，即：

• 状态必须包含消息列表。
• 最后一条消息必须是 AIMessage。
• AIMessage 必须填充 tool_calls。
  我们尝试进行一下实践。首先，既然是工具调用代理，我们就准备一下需要用的外部工具 / 函数。这里我们使用 Serper API 去构建实时联网检索的功能。

测试google 浏览器api方法的连通性

import requests
import json

from dotenv import load_dotenv
load_dotenv()
import os
weather_key = os.environ["OPEN_WEATHER_KEY"]
GOOGLE_SEARCH_KEY = os.environ["GOOGLE_SEARCH_KEY"]

def fetch_real_time_info(query):
    """Get real-time Internet information"""
    url = "https://google.serper.dev/search"
    payload = json.dumps({
        "q": query,
        "num": 1,
    })
    headers = {
        'X-API-KEY': GOOGLE_SEARCH_KEY,
        'Content-Type': 'application/json'
    }
    response = requests.post(url, headers=headers, data=payload)
    data = json.loads(response.text)  # 将返回的JSON字符串转换为字典
    if 'organic' in data:
        return json.dumps(data['organic'], ensure_ascii=False)  # 返回'organic'部分的JSON字符串
    else:
        return json.dumps({"error": "No organic results found"}, ensure_ascii=False)  # 如果没有'organic'键，返回错误信息

# 使用示例
query = "小米汽车"
result = fetch_real_time_info(query)
print(f"{"连通性测试："}\n{'*'*80}\n{result}\n{'*'*80}")

ToolNode

如果功能正常，该函数将根据用户的输入，返回实时的网页检索信息，这包括标题、链接、摘要等等有效的信息。而如果想要将普通的函数变成 ToolNode 可以应用的外部函数，只需要在函数定义时添加 @tool 装饰器。

单工具测试

from langchain_core.tools import tool
import requests
import json
# 如果功能正常，该函数将根据用户的输入，返回实时的网页检索信息，这包括标题、链接、摘要等等有效的信息。而如果想要将普通的函数变成 ToolNode 可以应用的外部函数，只需要在函数定义时添加 @tool 装饰器。
from dotenv import load_dotenv
load_dotenv()
import os
weather_key = os.environ["OPEN_WEATHER_KEY"]
GOOGLE_SEARCH_KEY = os.environ["GOOGLE_SEARCH_KEY"]
@tool
def fetch_real_time_info(query):
    """Get real-time Internet information"""
    url = "https://google.serper.dev/search"
    payload = json.dumps({
        "q": query,
        "num": 1,
    })
    headers = {
        'X-API-KEY': GOOGLE_SEARCH_KEY,
        'Content-Type': 'application/json'
    }
    response = requests.post(url, headers=headers, data=payload)
    data = json.loads(response.text)  # 将返回的JSON字符串转换为字典
    if 'organic' in data:
        return json.dumps(data['organic'], ensure_ascii=False)  # 返回'organic'部分的JSON字符串
    else:
        return json.dumps({"error": "No organic results found"}, ensure_ascii=False)  # 如果没有'organic'键，返回错误信息

# 导入预构建工具执行节点
from langgraph.prebuilt import ToolNode

# 存放所有已用@tool装饰的工具列表
tools = [fetch_real_time_info]
# 实例化ToolNode工具执行节点
tool_node = ToolNode(tools)

# ToolNode 使用消息列表对图状态进行操作。所以它要求消息列表中的最后一条消息是带有 tool_calls 参数的 AIMessage，
# 比如我们可以手动调用工具节点（模拟调用工具）：

from langchain_core.messages import AIMessage

message_with_single_tool_call = AIMessage(
    content="",
    tool_calls=[
        {
            "name": "fetch_real_time_info",
            "args": {"query": "小米汽车"},
            "id": "tool_call_id",
            "type": "tool_call",
        }
    ],
)

# tool_node.invoke({"messages": [message_with_single_tool_call]})

from langgraph.runtime import Runtime
from langgraph._internal._constants import CONF, CONFIG_KEY_RUNTIME
result = tool_node.invoke(
    {"messages": [message_with_single_tool_call]},
    config={CONF: {CONFIG_KEY_RUNTIME: Runtime()}},)

print(f"{"模拟调用结果："}\n{'*'*80}\n{result}\n{'*'*80}")

再增加一个工具，模拟将多个工具调用同时传递给 AIMessage 的 tool_calls 参数，仍然可以使用 ToolNode 进行并行工具调用：
模拟多工具调用：

from langchain_core.tools import tool
import requests
import json
# 如果功能正常，该函数将根据用户的输入，返回实时的网页检索信息，这包括标题、链接、摘要等等有效的信息。而如果想要将普通的函数变成 ToolNode 可以应用的外部函数，只需要在函数定义时添加 @tool 装饰器。
from dotenv import load_dotenv
load_dotenv()
import os
weather_key = os.environ["OPEN_WEATHER_KEY"]
GOOGLE_SEARCH_KEY = os.environ["GOOGLE_SEARCH_KEY"]

@tool
def fetch_real_time_info(query):
    """Get real-time Internet information"""
    url = "https://google.serper.dev/search"
    payload = json.dumps({
        "q": query,
        "num": 1,
    })
    headers = {
        'X-API-KEY': GOOGLE_SEARCH_KEY,
        'Content-Type': 'application/json'
    }
    response = requests.post(url, headers=headers, data=payload)
    data = json.loads(response.text)  # 将返回的JSON字符串转换为字典
    if 'organic' in data:
        return json.dumps(data['organic'], ensure_ascii=False)  # 返回'organic'部分的JSON字符串
    else:
        return json.dumps({"error": "No organic results found"}, ensure_ascii=False)  # 如果没有'organic'键，返回错误信息

@tool
def get_weather(location):
    """Call to get the current weather."""
    if location.lower() in ["beijing"]:
        return "北京的温度是16度，天气晴朗。"
    elif location.lower() in ["shanghai"]:
        return "上海的温度是20度，部分多云。"
    else:
        return "不好意思，并未查询到具体的天气信息。"

# 导入预构建工具执行节点
from langgraph.prebuilt import ToolNode

# 存放所有已用@tool装饰的工具列表
tools = [fetch_real_time_info, get_weather]
# 实例化ToolNode工具执行节点
tool_node = ToolNode(tools)
from langchain_core.messages import AIMessage
message_with_multiple_tool_calls = AIMessage(
    content="",
    tool_calls=[
        {
            "name": "fetch_real_time_info",
            "args": {"query": "小米汽车"},
            "id": "tool_call_id",
            "type": "tool_call",
        },
        {
            "name": "get_weather",
            "args": {"location": "beijing"},
            "id": "tool_call_id_2",
            "type": "tool_call",
        },
    ],
)

from langgraph.runtime import Runtime
from langgraph._internal._constants import CONF, CONFIG_KEY_RUNTIME
result = tool_node.invoke(
    {"messages": [message_with_multiple_tool_calls]},
    config={CONF: {CONFIG_KEY_RUNTIME: Runtime()}},)

print(f"{"模拟将多个工具调用同时传递给 AIMessage："}\n{'*'*80}\n{result}\n{'*'*80}")

# 而 Tool Calling Agent 的本质原理是：让大模型根据用户的输入，自动的去判断应该使用哪个函数，并实际的执行，最后结合工具的响应结果，用户的原始问题作为完整的 Prompt 生成最终的问题。
# 用大模型实现工具路由调用

# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=10)
from langchain_core.prompts import ChatPromptTemplate

result = llm.bind_tools(tools)

print(f"{"绑定大模型之后的工具："}\n{'*'*80}\n{result}\n{'*'*80}")

print(f"{"工具方面的详细信息："}\n{'*'*80}\n{result.kwargs}\n{'*'*80}")

print(f"{"测试调用工具,直接看tool_call里面是不是正确的工具："}\n{'*'*80}\n{result.invoke("小米汽车").tool_calls}\n{'*'*80}")

print(f"{"测试调用工具,直接看tool_call里面是不是正确的工具："}\n{'*'*80}\n{result.invoke("北京现在天气怎么样？").tool_calls}\n{'*'*80}")

# 由此可见，借助大模型可以正确的填充 tool_calls 信息，因此我们可以将其直接传递给 ToolNode，从而完成完整的 Tool Calling Agent 链路。代码如下：
result_validation  = tool_node.invoke({"messages": [result.invoke("小米汽车")]},
    config={CONF: {CONFIG_KEY_RUNTIME: Runtime()}},)

print(f"{"Tool Calling Agent 链路测试1："}\n{'*'*80}\n{result_validation}\n{'*'*80}")

result  = tool_node.invoke({"messages": [result.invoke("beijing现在天气怎么样？")]},
    config={CONF: {CONFIG_KEY_RUNTIME: Runtime()}},)

print(f"{"Tool Calling Agent 链路测试2："}\n{'*'*80}\n{result}\n{'*'*80}")

III Tool Calling Agent 的完整实现案例

接下来，我们来构建完整的 Tool Calling Agent。这里我们对 Router Agent 实现的图做进一步的升级，即用户输入问题后，如果不需要外部工具的信息，则直接生成回复，否则，则进入一个工具库中，选择最合适的工具执行，并返回最终的响应。因此，我们首先来定义工具库：

from pydantic import BaseModel, Field  
class SearchQuery(BaseModel):
    query: str = Field(description="Questions for networking queries")

# 5、工具定义：
from langchain_core.tools import tool
import requests
import json
from dotenv import load_dotenv
load_dotenv()
import os
weather_key = os.environ["OPEN_WEATHER_KEY"]
GOOGLE_SEARCH_KEY = os.environ["GOOGLE_SEARCH_KEY"]
@tool(args_schema = SearchQuery)
def fetch_real_time_info(query):
    """Get real-time Internet information"""
    url = "https://google.serper.dev/search"
    payload = json.dumps({
        "q": query,
        "num": 1,
    })
    headers = {
        'X-API-KEY': GOOGLE_SEARCH_KEY,
        'Content-Type': 'application/json'
    }
    response = requests.post(url, headers=headers, data=payload)
    data = json.loads(response.text)  # 将返回的JSON字符串转换为字典
    if 'organic' in data:
        return json.dumps(data['organic'], ensure_ascii=False)  # 返回'organic'部分的JSON字符串
    else:
        return json.dumps({"error": "No organic results found"}, ensure_ascii=False)  # 如果没有'organic'键，返回错误信息

# 工具1：天气查询工具
class WeatherLoc(BaseModel):
    location: str = Field(description="The location name of the city")

@tool(args_schema = WeatherLoc)
def get_weather(location):
    """Call to get the current weather."""
    if location.lower() in ["beijing"]:
        return "北京的温度是16度，天气晴朗。"
    elif location.lower() in ["shanghai"]:
        return "上海的温度是20度，部分多云。"
    else:
        return "不好意思，并未查询到具体的天气信息。"

from typing import Union
# 工具2：数据库插入用户信息工具
from typing import Optional, Literal
from pydantic import BaseModel, Field   
class UserInfo(BaseModel):
    """Extracted user information, such as name, age, email, and phone number, if relevant."""
    name: str = Field(description="The name of the user")
    age: Optional[int] = Field(description="The age of the user")
    email: str = Field(description="The email address of the user")
    # phone: Optional[str] = Field(description="The phone number of the user")
    phone: Optional[Union[str, int]] = Field(description="The phone number of the user")

from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData
from sqlalchemy.orm import declarative_base, sessionmaker
passwrod = os.environ["MYSQL_PASSWORD"]
acc = os.environ["MYSQL_account"]
DATABASE_URI = f'mysql+pymysql://{acc}:{passwrod}@localhost/langgraph?charset=utf8mb4'
engine = create_engine(DATABASE_URI, echo=True)
# 创建基类
Base = declarative_base()
# 自动创建数据表（不存在则新建，存在不会覆盖）
Base.metadata.create_all(engine)

# 创建数据库会话工厂
Session = sessionmaker(bind=engine)
session = Session()

# 定义 UserInfo 映射模型，对应数据库 users 表
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    age = Column(Integer)
    email = Column(String(100))
    phone = Column(String(15))

@tool(args_schema = UserInfo)
def insert_db(name, age, email, phone):
    """Insert user information into the database, The required parameters are name, age, email, phone"""
    session = Session()  # 确保为每次操作创建新的会话
    try:
        # 创建用户实例
        user = User(name=name, age=age, email=email, phone=phone)
        # 添加到会话
        session.add(user)
        # 提交事务
        session.commit()
        return {"messages": ["数据已成功存储至Mysql数据库。"]}
    except Exception as e:
        session.rollback()  # 出错时回滚
        return {"messages": [f"数据存储失败，错误原因：{e}"]}
    finally:
        session.close()  # 关闭会话

# 三个工具定义完毕后，我们将其传递给 ToolNode，并进行大模型的实例化。
# 导入预构建工具执行节点
from langgraph.prebuilt import ToolNode
tools = [insert_db, fetch_real_time_info, get_weather]
tool_node = ToolNode(tools)
print(f"{"ToolNode："}\n{'*'*80}\n{tool_node}\n{'*'*80}")

# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=20)

# 这里仍然使用 Pydantic 来做结构化输出，帮助 Router Function 来选择路径分支。

# 定义正常生成模型回复的模型
class ConversationalResponse(BaseModel):
    """Respond to the user's query in a conversational manner. Be kind and helpful."""
    response: str = Field(description="A conversational response to the user's query")

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

class FinalResponse(BaseModel):
    """先判断用户消息属于哪一类，再填对应字段。
    判定规则：
    - 含姓名/年龄/邮箱/电话/手机号等个人信息（我叫/我是/今年X岁/邮箱/手机）→ route="insert_db"，并填 insert_db
    - 询问某地天气（今天天气/北京天气）→ route="get_weather"，并填 get_weather
    - 需要联网获取实时信息（小米汽车最新款/某新闻）→ route="fetch_real_time_info"，并填 fetch_real_time_info
    - 纯寒暄/闲聊/解释类提问（你好/谢谢/解释一下XX）→ route="chat"，并填 chat"""
    route: Literal["insert_db", "get_weather", "fetch_real_time_info", "chat"] = Field(
        description="消息类型路由：个人信息→insert_db，天气→get_weather，联网搜索→fetch_real_time_info，纯对话→chat"
    )
    insert_db: Optional[UserInfo] = Field(default=None, description="route=insert_db 时填写")
    get_weather: Optional[WeatherLoc] = Field(default=None, description="route=get_weather 时填写")
    fetch_real_time_info: Optional[SearchQuery] = Field(default=None, description="route=fetch_real_time_info 时填写")
    chat: Optional[ConversationalResponse] = Field(default=None, description="route=chat 时填写")

    @property
    def final_output(self):
        """下游通过 .final_output 访问；按 route 返回当前选中的子对象，
        使其与原 Union 设计兼容（isinstance 判断、.response 访问、str() 喂模型都不受影响）。"""
        return getattr(self, self.route)

# 依次定义三个节点函数：

# def chat_with_model(state):
#     """generate structured output"""
#     # print(state)
#     # print("------------------")
#     messages = state['messages']
#     structured_llm = llm.with_structured_output(FinalResponse)
#     response = structured_llm.invoke(messages)
#     return {"messages": [response]}
ROUTER_SYSTEM_PROMPT = """
你是一个严格的消息路由分类器。你的唯一任务是：判断用户当前消息属于下面哪一类，并只填写对应字段，其余字段必须为 null。不要回答用户问题，不要解释，只做分类。

【四条路由的精确定义与边界】

1. route="insert_db" —— 用户在主动提供自己的个人信息，目的是"登记/录入/保存"。
   触发信号（满足任一即可，但必须是"在告知自己的信息"）：
   - 含自指词："我叫/我是/我的名字是/本人/我姓"
   - 含年龄自述："今年X岁/我X岁了"
   - 含联系方式："邮箱是/电话是/手机是/我的邮箱/我的电话"
   典型："我叫张三，今年20，邮箱z@163.com，电话138xxx"
   边界：用户只是"问别人的信息"（"张三的邮箱是多少"）不算，归 chat。
   填写：insert_db 字段，把识别到的 name/age/email/phone 填进去，缺的填 null。

2. route="get_weather" —— 用户在询问某地的天气状况。
   触发信号：含"天气/气温/多少度/下雨吗/冷不冷/热不热"且指向一个地点。
   典型："北京天气怎么样/上海今天多少度/beijing的天气"
   边界："今天天气真好啊"（陈述感受，非查询）不算，归 chat。
   填写：get_weather 字段，location 填城市名（中文或英文统一转小写英文，如"北京"→"beijing"）。

3. route="fetch_real_time_info" —— 用户在询问需要联网才能回答的实时/外部信息。
   触发信号（满足任一）：
   - 具体品牌/产品/新闻/人物/事件的最新情况："小米汽车/苹果发布会/某某公司最新"
   - 需要"当前/最新/现在/最近"等时效词的事实查询
   - 任何你训练数据里没有、必须上网查的内容
   典型："小米汽车最新款是什么/最近有什么新闻/OpenAI最新模型"
   边界：常识性问题（"Python是什么"）不算，归 chat；天气不算（走 get_weather）。
   填写：fetch_real_time_info 字段，query 填一个适合搜索引擎的简洁查询词。

4. route="chat" —— 以上三类都不满足时的兜底。
   包括：寒暄（你好/谢谢/再见）、闲聊、知识问答、解释概念、对上一轮的追问、无法判断意图。
   典型："你好/解释一下什么是RAG/谢谢你/今天心情不好"
   填写：chat 字段，response 填你对用户的自然语言回复。

【优先级与冲突处理】
当多条规则似乎都命中时，按此优先级判定：
  insert_db（主动给个人信息）> get_weather（明确问天气）> fetch_real_time_info（明确要联网）> chat
即：用户"在提供自己的信息"优先于一切；其次明确问天气；其次明确要联网；都不明确才 chat。
判断不准时，宁可选 chat，不要把闲聊误判成工具调用。

【输出要求】
- route 必须是上述四个字面量之一，不得自造。
- 只填 route 对应的那一个字段，另外三个字段必须为 null。
- 不要输出任何额外文字，不要在字段里塞 JSON 字符串。"""

def chat_with_model(state):
    messages = state['messages']
    structured_llm = llm.with_structured_output(FinalResponse)
    # 在用户消息前注入路由系统提示
    response = structured_llm.invoke([SystemMessage(content=ROUTER_SYSTEM_PROMPT)] + messages)
    return {"messages": [response]}

def final_answer(state):
    """generate natural language responses"""
    # print(state)
    # print("------------------")
    messages = state['messages'][-1]
    response = messages.final_output.response
    return {"messages": [response]}
model_with_tools = llm.bind_tools(tools)

def execute_function(state):
    """generate natural language responses"""
    messages = state['messages'][-1].final_output
    response = tool_node.invoke({"messages": [model_with_tools.invoke(str(messages))]})
    print(f"response:{response}")
    response = response["messages"][0].content
    return {"messages": [response]}

# 定义好了所有节点函数后，开始构建图。
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
# 定义图的状态模式。
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

# 定义 generate_branch 函数作为 Router Function，根据经过 chat_with_model 节点后产生的不同 Pydantic 对象，选择连接不同的节点，即 final_answer 或 execute_function。

def generate_branch(state: AgentState):
    result = state['messages'][-1]
    output = result.final_output
    if isinstance(output, ConversationalResponse):
        return False
    else:
        return True

# 构建图并使用条件边来生成 Router。代码如下：

graph = StateGraph(AgentState)

# 添加三个节点
graph.add_node("chat_with_model", chat_with_model)
graph.add_node("final_answer", final_answer)
graph.add_node("execute_function", execute_function)

# 设置图的启动节点
graph.set_entry_point("chat_with_model")

# 设置条件边
graph.add_conditional_edges(
    "chat_with_model",
    generate_branch,
    {True: "execute_function", False: "final_answer"}
)

# 设置终止节点
graph.set_finish_point("final_answer")
graph.set_finish_point("execute_function")

# 编译图
graph = graph.compile()

image_data = graph.get_graph().draw_mermaid_png()
with open("Router_Agent_G.png", "wb") as f:
    f.write(image_data)

# 完成图的编译后，便可以开始进行功能测试，这里我们选择不同的输入场景，以测试是否可以正确的进入到指定的功能函数中完成需求。
# 场景 1：调用对话工具
query1="你好，请你介绍一下你自己"
input_message1 = {"messages": [HumanMessage(content=query1)]}

result1 = graph.invoke(input_message1)
print(f"{"纯对话测试，结构化输出："}\n{'*'*80}\n{result1}\n{'*'*80}\n")
# 查看最终输出
result1 = result1["messages"][-1]
print(f"{"纯对话测试，只看AI响应的原话："}\n{'*'*80}\n{result1}\n{'*'*80}\n")

# 场景 2：调用谷歌浏览器查询工具
query2="小米汽车"
input_message2 = {"messages": [HumanMessage(content=query2)]}

result2 = graph.invoke(input_message2)
print(f"{"需要联网搜索工具，结构化输出："}\n{'*'*80}\n{result2}\n{'*'*80}\n")

# 查看最终输出
result2 = result2["messages"][-1]
print(f"{"需要联网搜索工具，只看AI响应的原话："}\n{'*'*80}\n{result2}\n{'*'*80}\n")

# 场景 3：调用天气查询工具

query3="shanghai的天气怎么样？"
input_message3 = {"messages": [HumanMessage(content=query3)]}

result3 = graph.invoke(input_message3)
print(f"{"调用天气查询工具，结构化输出："}\n{'*'*80}\n{result3}\n{'*'*80}\n")

# 查看最终输出
result3 = result3["messages"][-1]
print(f"{"调用天气查询工具，只看AI响应的原话："}\n{'*'*80}\n{result3}\n{'*'*80}\n")

# 场景 4：调用数据库插入用户信息工具
query4="我叫朱迪，大明成祖皇帝，今年278岁，邮箱是judy@163.com，电话是110110110110110"
input_message4 = {"messages": [HumanMessage(content=query4)]}

result4 = graph.invoke(input_message4)
print(f"{"调用数据库插入用户信息工具，结构化输出："}\n{'*'*80}\n{result4}\n{'*'*80}\n")

IV 手动构建 Tool Calling Agent 的方法

当然，工具调用的过程也可以手动进行实现，我们只需要进行适当的逻辑修改即可，如下示例所示：

# 1、获取google浏览器工具定义：
from pydantic import BaseModel, Field  
class SearchQuery(BaseModel):
    query: str = Field(description="Questions for networking queries")

from langchain_core.tools import tool
import requests
import json
from dotenv import load_dotenv
load_dotenv()
import os
weather_key = os.environ["OPEN_WEATHER_KEY"]
GOOGLE_SEARCH_KEY = os.environ["GOOGLE_SEARCH_KEY"]
@tool(args_schema = SearchQuery)
def fetch_real_time_info(query):
    """Get real-time Internet information"""
    url = "https://google.serper.dev/search"
    payload = json.dumps({
        "q": query,
        "num": 1,
    })
    headers = {
        'X-API-KEY': GOOGLE_SEARCH_KEY,
        'Content-Type': 'application/json'
    }
    response = requests.post(url, headers=headers, data=payload)
    data = json.loads(response.text)  # 将返回的JSON字符串转换为字典
    if 'organic' in data:
        return json.dumps(data['organic'], ensure_ascii=False)  # 返回'organic'部分的JSON字符串
    else:
        return json.dumps({"error": "No organic results found"}, ensure_ascii=False)  # 如果没有'organic'键，返回错误信息

# 2、插入mysql数据库
from typing import Union
from typing import Optional, Literal
from pydantic import BaseModel, Field   
class UserInfo(BaseModel):
    """Extracted user information, such as name, age, email, and phone number, if relevant."""
    name: str = Field(description="The name of the user")
    age: Optional[int] = Field(description="The age of the user")
    email: str = Field(description="The email address of the user")
    phone: Optional[Union[str, int]] = Field(description="The phone number of the user")

from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData
from sqlalchemy.orm import declarative_base, sessionmaker
passwrod = os.environ["MYSQL_PASSWORD"]
acc = os.environ["MYSQL_account"]
DATABASE_URI = f'mysql+pymysql://{acc}:{passwrod}@localhost/langgraph?charset=utf8mb4'
engine = create_engine(DATABASE_URI)
# 创建基类
Base = declarative_base()
# 自动创建数据表（不存在则新建，存在不会覆盖）
Base.metadata.create_all(engine)

# 创建数据库会话工厂
Session = sessionmaker(bind=engine)
session = Session()

# 定义 UserInfo 映射模型，对应数据库 users 表
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    age = Column(Integer)
    email = Column(String(100))
    phone = Column(String(15))

@tool(args_schema = UserInfo)
def insert_db(name, age, email, phone):
    """Insert user information into the database, The required parameters are name, age, email, phone"""
    session = Session()  # 确保为每次操作创建新的会话
    try:
        # 创建用户实例
        user = User(name=name, age=age, email=email, phone=phone)
        # 添加到会话
        session.add(user)
        # 提交事务
        session.commit()
        return {"messages": ["数据已成功存储至Mysql数据库。"]}
    except Exception as e:
        session.rollback()  # 出错时回滚
        return {"messages": [f"数据存储失败，错误原因：{e}"]}
    finally:
        session.close()  # 关闭会话

# 工具3：天气查询工具
class WeatherLoc(BaseModel):
    location: str = Field(description="The location name of the city")

@tool(args_schema = WeatherLoc)
def get_weather(location):
    """Call to get the current weather."""
    if location.lower() in ["beijing"]:
        return "北京的温度是16度，天气晴朗。"
    elif location.lower() in ["shanghai"]:
        return "上海的温度是20度，部分多云。"
    else:
        return "不好意思，并未查询到具体的天气信息。"

# 测试连通性
print(f'''
name: {fetch_real_time_info.name}
description: {fetch_real_time_info.description}
arguments: {fetch_real_time_info.args}
''')

# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=10)

def chat_with_model(state):
    """generate structured output"""
    messages = state['messages']
    response = llm.invoke(messages)  # 这里可以不使用格式化输出
    return {"messages": [response]}

# 2、执行工具节点。
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
# 定义图的状态模式。
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
def execute_function(state: AgentState):
    tool_calls = state['messages'][-1].tool_calls
    results = []
    tools = [insert_db, fetch_real_time_info, get_weather]
    tools = {t.name: t for t in tools}
    for t in tool_calls:
        if not t['name'] in tools:
            result = "bad tool name, retry"
        else:
            result = tools[t['name']].invoke(t['args'])
        results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
    return {'messages': results}

def final_answer(state):
    """generate natural language responses"""
    messages = state['messages'][-1]
    return {"messages": [messages]}

# 请你基于现在得到的信息，进行总结，生成专业的回复，注意，请用中文回复
SYSTEM_PROMPT = """
Please summarize the information obtained so far and generate a professional response. Note, please reply in Chinese.
"""

# 生成最终自然语言回复节点
def natural_response(state):
    """generate final language responses"""
    messages = state['messages'][-1]
    messages = [SystemMessage(content=SYSTEM_PROMPT)] + [HumanMessage(content=messages.content)]
    response = llm.invoke(messages)
    return {"messages": [response]}

# 路由判断：是否存在工具调用
def exists_function_calling(state: AgentState):
    result = state['messages'][-1]
    return len(result.tool_calls) > 0

# 图状态定义
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

from IPython.display import Image, display
from langgraph.graph import StateGraph

# 初始化图
graph = StateGraph(AgentState)

# 注册全部节点
graph.add_node("chat_with_model", chat_with_model)
graph.add_node("execute_function", execute_function)
graph.add_node("final_answer", final_answer)
graph.add_node("natural_response", natural_response)

# 入口节点
graph.set_entry_point("chat_with_model")

# 条件分支：模型输出是否带工具调用
graph.add_conditional_edges(
    "chat_with_model",
    exists_function_calling,
    {True: "execute_function", False: "final_answer"}
)

# 固定边：工具执行/直接回答后统一汇总生成自然语言
graph.add_edge("execute_function", "natural_response")
graph.add_edge("final_answer", "natural_response")

# 终止节点
graph.set_finish_point("natural_response")

# 编译流程图
graph = graph.compile()

# 可视化流程图
# image_data = graph.get_graph().draw_mermaid_png()
# with open("man-made_Tool_Calling_Agent_G.png", "wb") as f:
#     f.write(image_data)

# 一、绑定工具到 LLM
# 工具列表
tools = [insert_db, fetch_real_time_info, get_weather]
# 将工具绑定到大模型，模型输出会携带tool_calls
llm = llm.bind_tools(tools)

# 二、测试用例 1：纯闲聊，无需调用工具
query = "你好，请你介绍一下你自己"
input_message = {"messages": [HumanMessage(content=query)]}
result = graph.invoke(input_message)
# 打印最终AI自然语言回复
result=result["messages"][-1].content
print(f"{"测试用例 1：纯闲聊，无需调用工具："}\n{'*'*80}\n{result}\n{'*'*80}\n\n")

# 测试用例 2：联网搜索工具（fetch_real_time_info）
query = "小米汽车"
input_message = {"messages": [HumanMessage(content=query)]}
result = graph.invoke(input_message)
result=result["messages"][-1].content
print(f"{"测试用例 2：联网搜索工具："}\n{'*'*80}\n{result}\n{'*'*80}\n\n")

# 四、测试用例 3：用户信息入库工具（insert_db）
query = "我叫艾斯奥特曼，今年38岁，邮箱地址是aoteman#qq.com，电话是12111111111111"
input_message = {"messages": [HumanMessage(content=query)]}
result = graph.invoke(input_message)
result=result["messages"][-1].content
print(f"{"测试用例 3：用户信息入库工具："}\n{'*'*80}\n{result}\n{'*'*80}\n\n")

以上就是两个基于 LangGraph 实现的 Tool Calling Agent 代理架构，从整体上看，无论是对于图结构，还是工具的接入，可扩展性都非常高，使用这个代理架构需要注意的是 Router Function 的分支判断，除此之外，对于工具调用而言，因为调用工具的参数是由大模型根据用户输入的自然语言生成的，所以一定会存在尝试调用不存在的工具，或者无法返回与请求的架构匹配的参数等边缘情况，会直接导致整个图的运行中断。这一部分进阶优化的内容，我们在下一小节再进行详细的介绍和实践。

四、自主循环代理(React Agent)

LangGraph 中 ReAct 的构建原理

LangGraph 内置预构建组件介绍

LangGraph 提供开箱即用的预制组件：
ToolNode：上一节讲解过，专门处理外部工具调用，底层封装图结构，接收 JSON Schema 格式参数，执行工具并返回结果。
ReAct 代理架构：官方封装好的循环智能代理，和手动实现 ReAct 的核心思路完全一致，但做了图结构适配

LangGraph ReAct 循环执行逻辑

本质是while 自治循环：
大模型执行推理，自主判断是否需要调用工具、选择工具、生成工具入参；
执行选中的工具，拿到工具返回观测结果；
将工具输出放回上下文，再次送入大模型；
模型判断无需再调用工具时，终止循环，输出最终自然语言答案

底层核心构成

LangGraph 内置 ReAct 不是黑盒，底层由三部分组合成一张可循环图：
LLM（绑定工具的 Tool Calling 大模型）
Tool Calling 工具执行节点（ToolNode）
Router 路由判断节点（区分「继续调用工具」/「结束输出」）
整体是自治循环代理图，开箱即用，无需手动写节点、分支、循环逻辑。

基于 LangGraph 内置 ReAct 组件，搭建天气查询智能助手。

I 简单实现代码

# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=10)

from langchain_core.tools import tool
from typing import Union, Optional
from pydantic import BaseModel, Field
import requests
import json
from dotenv import load_dotenv
load_dotenv()
import os
weather_key = os.environ["OPEN_WEATHER_KEY"]
class WeatherLoc(BaseModel):
    location: str = Field(description="The location name of the city")
# 工具3：天气查询工具
@tool(args_schema=WeatherLoc)
def get_weather(location):
    """
    Function to query current weather.
    :param loc: Required parameter, of type string, representing the specific city name for the weather query. \
    Note that for cities in China, the corresponding English city name should be used. For example, to query the weather for Beijing, \
    the loc parameter should be input as 'Beijing'.
    :return: The result of the OpenWeather API query for current weather, with the specific URL request address being: https://api.openweathermap.org/data/2.5/weather
    The return type is a JSON-formatted object after parsing, represented as a string, containing all important weather information.
    """
    # Step 1. 构建请求
    url = "https://api.openweathermap.org/data/2.5/weather"

    # Step 2. 设置查询参数
    params = {
        "q": location,
        "appid": weather_key,   # 输入API key
        "units": "metric",                           # 使用摄氏度而不是华氏度
        "lang":"zh_cn"                               # 输出语言为简体中文
    }

    # Step 3. 发送GET请求
    response = requests.get(url, params=params)

    # Step 4. 解析响应
    data = response.json()
    return json.dumps(data)

result = get_weather.invoke({"location": "TOKYO"}) 
# print(f"{"测试调用天气工具："}\n{'*'*80}\n{result}\n{'*'*80}\n\n")

tools = [get_weather]

from langchain.agents import create_agent
from langchain_core.messages import HumanMessage

# 系统提示词：指导 Agent 的行为
# system_prompt = "你是一个有用的天气助手。请根据用户的问题，使用工具查询天气信息，并用中文给出简洁友好的回答。"

prompt = """Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought:{agent_scratchpad}

"""

from langchain.agents import create_agent
from langchain_core.messages import HumanMessage

# 创建 Agent（新版 API）
agent = create_agent(llm, tools, system_prompt=prompt)

# ── 示例 1：单城市 ──
# print("=" * 70)
# for chunk in agent.stream(
#     {"messages": [HumanMessage(content="大理今天的天气是多少度？")]},
#     stream_mode="values"
# ):
#     print(f"\n{"换行"}\n{"换行"}")
#     chunk["messages"][-1].pretty_print()   # 框架自带，一行搞定 ReAct 展示

result = agent.invoke({"messages": [HumanMessage(content="大理今天的天气是多少度？")]},)
# print(result)
# 只打印messages字段中的AIMessage字段里的回答内容
result["messages"][-1].pretty_print()
# ── 示例 2：多城市对比 ──
# print("\n" + "=" * 70)
# for chunk in agent.stream(
#     {"messages": [HumanMessage(content="现在昆明和大理今天的天气差多少度？")]},
#     stream_mode="values"
# ):
#     print("\n换行\n换行")
#     chunk["messages"][-1].pretty_print()

result = agent.invoke({"messages": [HumanMessage(content="现在昆明和大理今天的天气差多少度？")]},)
# print(result)
# 只打印messages字段中的AIMessage字段里的回答内容
result["messages"][-1].pretty_print()

2. 案例实操：构建复杂工具应用的 ReAct 自治代理

在这个案例中，我们将通过一个多工具场景需求来测试 LangGraph 中 ReAct 代理的构建方法和效果。我们设计了几个工具，以实现实时数据的查询和管理。
首先，用户可以通过一个工具根据城市名称实时获取当前天气信息。接着，如果用户希望将查询到的天气数据保存到本地数据库中，可以使用另一个工具完成数据的插入操作。
此外，我们还提供了一个工具，允许用户基于本地数据库中的天气数据进行提问数据进行提问。
通过这些工具的组合，我们能够快速验证如何在复杂的应用场景中有效地整合不同功能，并实际的感知 LangGraph 框架下 ReAct 代理模式带来的开发便捷性和可扩展性。
首先，我们接入实时天气数据查询的在线 API，代码定义如下：
OpenWeather API 的注册与使用，注册地址：https://openweathermap.org/

# 工具3：天气查询工具
from langchain_core.tools import tool
from typing import Union, Optional
from pydantic import BaseModel, Field
import requests
import json
from dotenv import load_dotenv
load_dotenv()
import os
weather_key = os.environ["OPEN_WEATHER_KEY"]
class WeatherLoc(BaseModel):
    location: str = Field(description="The location name of the city")

class WeatherInfo(BaseModel):
    """Extracted weather information for a specific city."""
    city_id: int = Field(default=0, description="可选，城市ID，不传则自动忽略")
    city_name: str = Field(..., description="The name of the city")
    main_weather: str = Field(..., description="The main weather condition")
    description: str = Field(..., description="A detailed description of the weather")
    temperature: float = Field(..., description="Current temperature in Celsius")
    feels_like: float = Field(..., description="Feels-like temperature in Celsius")
    temp_min: float = Field(..., description="Minimum temperature in Celsius")
    temp_max: float = Field(..., description="Maximum temperature in Celsius")

# 工具1：天气查询工具
@tool(args_schema=WeatherLoc)
def get_weather(location):
    """
    Function to query current weather.
    :param loc: Required parameter, of type string, representing the specific city name for the weather query. \
    Note that for cities in China, the corresponding English city name should be used. For example, to query the weather for Beijing, \
    the loc parameter should be input as 'Beijing'.
    :return: The result of the OpenWeather API query for current weather, with the specific URL request address being: https://api.openweathermap.org/data/2.5/weather
    The return type is a JSON-formatted object after parsing, represented as a string, containing all important weather information.
    """
    # Step 1. 构建请求
    url = "https://api.openweathermap.org/data/2.5/weather"

    # Step 2. 设置查询参数
    params = {
        "q": location,
        "appid": weather_key,   # 输入API key
        "units": "metric",                           # 使用摄氏度而不是华氏度
        "lang":"zh_cn"                               # 输出语言为简体中文
    }

    # Step 3. 发送GET请求
    response = requests.get(url, params=params)

    # Step 4. 解析响应
    data = response.json()
    return json.dumps(data)

# 测试一下 get_weather 函数的有效性，正常情况下可以得到输入城市名的实时天气信息，测试代码如下：

# 接下来，设计一个用于存储实时天气信息的表。这里我们定义一个新的模型 Weather，并包括上述所提取出来的字段。连接 Mysql 数据库及创建表的代码如下所示：
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.orm import sessionmaker, declarative_base

# 创建基类
Base = declarative_base()

# 定义 WeatherInfo 模型
class Weather(Base):
    __tablename__ = 'weather'
    id = Column(Integer, primary_key=True, autoincrement=True)  # 新增自增主键
    city_id = Column(Integer, primary_key=True)  # 城市ID
    city_name = Column(String(50))                # 城市名称
    main_weather = Column(String(50))            # 主要天气状况
    description = Column(String(100))             # 描述
    temperature = Column(Float)                  # 温度
    feels_like = Column(Float)                   # 体感温度
    temp_min = Column(Float)                     # 最低温度
    temp_max = Column(Float)                     # 最高温度

"""
接下来，使用 LangChain 的 tool 装饰器将普通的函数注册为 LangGraph 中支持的工具服务，根据需求的设计，我们要依次创建三个外部函数，分别是：
get_weather 工具：用于根据城市名称实时查询该城市的当前天气数据。
insert_weather_to_db 工具：如果用户想要把查询到的天气数据插入到数据库的表中，则使用此函数完成数据库的插入操作。
query_weather_from_db 工具：如果用户想基于本地数据库的天气数据直接进行提问，则使用此函数完成数据库的查询操作。
如上节课实践的流程一样，我们依然使用 pydantic 来做工具的参数校验和结构化输出。三个工具函数的定义代码依次如下所示：
"""
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData
from sqlalchemy.orm import declarative_base, sessionmaker
passwrod = os.environ["MYSQL_PASSWORD"]
acc = os.environ["MYSQL_account"]
DATABASE_URI = f'mysql+pymysql://{acc}:{passwrod}@localhost/langgraph?charset=utf8mb4'
engine = create_engine(DATABASE_URI)

# 自动创建数据表（不存在则新建，存在不会覆盖）
Base.metadata.create_all(engine)

# 创建数据库会话工厂
Session = sessionmaker(bind=engine)

@tool(args_schema=WeatherInfo)
def insert_weather_to_db(city_id, city_name, main_weather, description, temperature, feels_like, temp_min, temp_max):
    """Insert weather information into the database."""
    session = Session() # 确保为每次操作创建新的会话
    try:
        # 创建天气实例
        weather = Weather(
            city_id=city_id,
            city_name=city_name,
            main_weather=main_weather,
            description=description,
            temperature=temperature,
            feels_like=feels_like,
            temp_min=temp_min,
            temp_max=temp_max
        )
        # 使用 merge 方法来插入或更新（如果已有记录则更新）
        session.merge(weather)
        # 提交事务
        session.commit()
        return {"messages": [f"天气数据已成功存储至Mysql数据库。"]}
    except Exception as e:
        session.rollback() # 出错时回滚
        return {"messages": [f"数据存储失败，错误原因：{e}"]}
    finally:
        session.close() # 关闭会话

class QueryWeatherSchema(BaseModel):
    """Schema for querying weather information by city name."""
    city_name: str = Field(..., description="The name of the city to query weather information")

@tool(args_schema=QueryWeatherSchema)
def query_weather_from_db(city_name: str):
    """Query weather information from the database by city name."""
    session = Session()
    try:
        # 查询天气数据
        weather_data = session.query(Weather).filter(Weather.city_name == city_name).first()
        if weather_data:
            return {
                "city_id": weather_data.city_id,
                "city_name": weather_data.city_name,
                "main_weather": weather_data.main_weather,
                "description": weather_data.description,
                "temperature": weather_data.temperature,
                "feels_like": weather_data.feels_like,
                "temp_min": weather_data.temp_min,
                "temp_max": weather_data.temp_max
            }
        else:
            return {"messages": [f"未找到城市 '{city_name}' 的天气信息。"]}
    except Exception as e:
        return {"messages": [f"查询失败，错误原因：{e}"]}
    finally:
        session.close() # 关闭会话

class SearchQuery(BaseModel):
    query: str = Field(description="Questions for networking queries")

from langchain_core.tools import tool
import requests
import json
from dotenv import load_dotenv
load_dotenv()
import os
weather_key = os.environ["OPEN_WEATHER_KEY"]
GOOGLE_SEARCH_KEY = os.environ["GOOGLE_SEARCH_KEY"]

@tool(args_schema = SearchQuery)
def fetch_real_time_info(query):
    """Get real-time Internet information"""
    url = "https://google.serper.dev/search"
    payload = json.dumps({
        "q": query,
        "num": 1,
    })
    headers = {
        'X-API-KEY': GOOGLE_SEARCH_KEY,
        'Content-Type': 'application/json'
    }
    response = requests.post(url, headers=headers, data=payload)
    data = json.loads(response.text)  # 将返回的JSON字符串转换为字典
    if 'organic' in data:
        return json.dumps(data['organic'], ensure_ascii=False)  # 返回'organic'部分的JSON字符串
    else:
        return json.dumps({"error": "No organic results found"}, ensure_ascii=False)  # 如果没有'organic'键，返回错误信息

tools = [fetch_real_time_info, get_weather, insert_weather_to_db, query_weather_from_db]
object = [i.name for i in tools]
print(f"{"工具列表："}\n{'*'*80}\n{object}\n{'*'*80}\n\n")

print(f"\n{'-'*80}\n{"以上将工具定义完毕"}\n{'-'*80}\n\n")

# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=120)

"""
当有了工具列表和模型后，就可以通过 create_react_agent 这个 LangGraph 框架中预构建的方法来创建自治循环代理（ReAct）的工作流，其必要的参数如下：
model：支持工具调用的 LangChain 聊天模型。
tools：工具列表、ToolExecutor 或 ToolNode 实例。
state_schema：图的状态模式。必须有 messages 和 is_last_step 键。默认为定义这两个键的 Agent State。
上述三点我们均在前面的课程中详细且作为重点介绍过，大家应该是比较容易理解的。所以，创建 ReAct 代理的代码就如下所示：
"""

from langgraph.prebuilt import create_react_agent

graph = create_react_agent(llm, tools=tools)

image_data = graph.get_graph().draw_mermaid_png()
with open("React_Agent_G.png", "wb") as f:
    f.write(image_data)

# 返回的是编译好的 LangGraph 可运行程序，可直接用于聊天交互。调用方式则和之前使用的方法一样，我们可以依次针对不同复杂程度的需求依次进行提问。首先是测试是否可以不使用工具，直接调用大模型生成响应。
# query="你好，请你介绍一下你自己"
# input_message = {"messages": [HumanMessage(content=query)]}

# 可以自动处理成 HumanMessage 的消息格式
finan_response = graph.invoke({"messages": ["你好，请你介绍一下你自己"]})

print(f"{"进行一般对话响应："}\n{'*'*80}\n{finan_response}\n{'*'*80}\n\n")

# finan_response = finan_response["messages"][-1].content
# print(f"{"进行一般对话响应："}\n{'*'*80}\n{finan_response}\n{'*'*80}\n\n")

# 加大输入问题的复杂度，接下来我们提问的问题希望它能够自动找到正确的工具函数，基于工具的执行结果作为既定的事实，引导生成最终的回复。

# finan_response = graph.invoke({"messages": ["吉林今天的天气怎么样？把查询的结果存到数据库"]})

# print(f"{"查询天气："}\n{'*'*80}\n{finan_response}\n{'*'*80}\n\n")
# print(f"{"查询天气调用了哪些工具1："}\n{'*'*80}\n{finan_response['messages'][1].tool_calls}\n{'*'*80}\n\n")
# # 遍历整个消息列表，提取所有包含 tool_calls 的消息
# all_tool_calls = [
#     msg.tool_calls 
#     for msg in finan_response['messages'] 
#     if hasattr(msg, 'tool_calls') and msg.tool_calls
# ]
# print(f"{"查询天气调用了哪些工具2："}\n{'*'*80}\n{all_tool_calls}\n{'*'*80}\n\n")

# finan_response = graph.invoke({"messages": ["去数据库查询一下吉林今天的天气怎么样？"]})  # 模拟用户输入：
# all_tool_calls = [
#     msg.tool_calls 
#     for msg in finan_response['messages'] 
#     if hasattr(msg, 'tool_calls') and msg.tool_calls
# ]
# print(f"{"查询天气："}\n{'*'*80}\n{finan_response}\n{'*'*80}\n\n")
# print(f"{"从数据库中查询天气调用了哪些工具："}\n{'*'*80}\n{all_tool_calls}\n{'*'*80}\n\n")

# finan_response = graph.invoke({"messages": ["你知道关于小米的销售情况吗？请用中文回复我"]})

# all_tool_calls = [
#     msg.tool_calls 
#     for msg in finan_response['messages'] 
#     if hasattr(msg, 'tool_calls') and msg.tool_calls
# ]
# print(f"{"查询天气："}\n{'*'*80}\n{finan_response}\n{'*'*80}\n\n")
# print(f"{"从数据库中查询天气调用了哪些工具："}\n{'*'*80}\n{all_tool_calls}\n{'*'*80}\n\n")

# 继续加大问题的难度，我们要在一个问题中涉及多个工具的使用，比如：

# finan_response = graph.invoke({"messages": ["帮我查一下北京、上海、哈尔滨三个城市的天气，告诉我哪个城市最适合出游。同时，把查询到的数据存储到数据库中"]})

# all_tool_calls = [
#     msg.tool_calls 
#     for msg in finan_response['messages'] 
#     if hasattr(msg, 'tool_calls') and msg.tool_calls
# ]
# print(f"{"查询天气："}\n{'*'*80}\n{finan_response}\n{'*'*80}\n\n")
# print(f"{"从数据库中查询天气调用了哪些工具："}\n{'*'*80}\n{all_tool_calls}\n{'*'*80}\n\n")

# 同时，可以在数据库中查看数据的插入情况：
finan_response = graph.invoke({"messages": ["帮我分析一下数据库中北京和哈尔滨城市天气的信息，做一个详细的对比，并生成出行建议"]})

all_tool_calls = [
    msg.tool_calls 
    for msg in finan_response['messages'] 
    if hasattr(msg, 'tool_calls') and msg.tool_calls
]
print(f"{"查询天气："}\n{'*'*80}\n{finan_response}\n{'*'*80}\n\n")
print(f"{"从数据库中查询天气调用了哪些工具："}\n{'*'*80}\n{all_tool_calls}\n{'*'*80}\n\n")