如何使用LangGraph开发AI智能体：一份实用指南

核心概念1：工具

可以把工具看作是AI智能体可以调用的Python函数。大语言模型会根据函数的名称、文档字符串、参数以及返回值来判断这些函数的功能及使用场景。

LangChain提供了一个@tool装饰器，可以让任何函数变成一个可使用的工具。例如：

from langchain_core.tools import tool

@tool
def get_current_weather(location: str) -> str:
    """获取指定地点的当前天气信息。
    
    当用户询问天气情况时可以使用这个函数。
    
    参数：
        location：城市名称（例如“San Francisco”或“New York”）
    
    返回值：
        天气描述字符串
    """
    # 在实际应用中，这里会调用天气API来获取数据
    return f"{location}的当前天气是晴朗的，气温为72华氏度"

注意，文档字符串本身就已经说明了这个函数的功能，大语言模型正是根据这些信息来判断是否应该使用这个函数的。

<这里有一个来自FinanceGPT的真实例子。FinanceGPT是一种用于查询财务交易记录的工具：

from langchain_core.tools import tool
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

def create_search_transactions_tool(search_space_id: int, db_session: AsyncSession):
    """
    这是一个工厂函数，用于创建能够访问数据库的搜索工具。

    这种设计模式允许你在不影响大语言模型接口清晰度的同时，注入所需的依赖项（如数据库连接和用户上下文信息）。
    """

    @tool
    async def search_transactions(
        keywords: str | None = None,
        category: str | None = None
    ) -> dict:
        """根据商家名称或消费类别搜索金融交易记录。

        当用户提出以下类型的问题时，可以使用这个函数：
        - 在特定商家处的消费金额（例如：“在星巴克花了多少钱？”）
        - 某个类别的消费金额（例如：“在食品杂货上花了多少钱？”）
        - 结合两种条件进行查询（例如：“显示我在麦当劳的餐饮消费记录”）

        参数：
            keywords：要搜索的商家名称
            category：消费类别（例如：“食品杂货”、“汽油”）

        返回值：
        一个字典，其中包含交易记录、总金额以及交易数量
        """

        # 查询数据库
        query = select(Document.document_metadata).where(
            Document.search_space_id == search_space_id
        )
        result = await db_session.execute(query)
        documents = result.all()

        # 根据指定条件筛选交易记录
        all_transactions = []
        for (docMetadata,) in documents:
            transactions = docmetadata.get("financial_data", "").get("transactions", [])
            
            for txn in transactions:
                # 应用过滤条件
                if category and category.lower() not in str(txn.get("category", "")).lower():
                    continue
                if keywords and keywords.lower() not in txn.get("description", "").lower():
                    continue
                
                # 将符合条件的交易记录添加到结果列表中
                all_transactions.append({
                    "date": txn.get("date"),
                    "description": txn.get("description"),
                    "amount": float(txn.get("amount", 0)),
                    "category": txn.get("category"),
                })
        
        # 计算总金额并返回结果
        total = sum(abs(t["amount"]) for t in all_transactions if t["amount"] < 0)
        
        return {
            "transactions": all_transactions[:20],  # 限制返回的结果数量
            "total_amount": total,
            "count": len(all_transactions),
            "summary": f"找到了{len(all_transactions)}笔交易记录，总金额为${total:,.2f}"
        }
    
    return search_transactions

让我们来详细了解一下这段代码的功能。

工厂函数模式：这个工具仅接收大语言模型能够提供的参数（即关键词和消费类别），但它还需要数据库会话以及search_space_id才能确定应该查询哪些数据。工厂函数通过将这些依赖项封装在闭包中来解决这个问题，这样一来，大语言模型看到的只是一个简洁的接口，而具体的数据库连接细节则被隐藏了起来。

过滤逻辑：我们会遍历所有交易记录，并应用可选的过滤条件。如果提供了category参数，那么该分类必须出现在交易记录的类别字段中；如果提供了keywords参数，那么这些关键词必须出现在商家的描述中。这两种条件可以同时使用，这样大型语言模型就能处理诸如“我在‘餐厅’类别下在麦当劳消费了多少钱？”这样的问题。

返回值：该工具并不会返回原始数据列表，而是会返回一个结构化的字典，其中包含有限数量的结果记录、预先计算出的总金额以及用通俗语言编写的摘要。通过这个摘要，大型语言模型可以直接读取“找到了23笔交易记录，总金额为1,245.67美元”这样的信息，从而立刻知道该如何回应用户，而无需自己去解析原始数据。

工具设计的核心原则

以下这些原则能够区分一个好的工具和一个出色的工具：

1. 详细的文档说明：文档中应对工具的功能进行详尽的解释，而不是给出模糊的描述。你提供的示例越多，大型语言模型就越容易选择合适的工具。

2. 简洁的设计结构：工具应该只接收那些大型语言模型能够访问和处理的参数。如果工具需要用户ID或数据库连接等信息，可以通过闭包将这些功能隐藏在工厂函数中。

3. 同时提供数据与摘要：除了原始数据外，如果还提供了摘要信息，使用者就能更容易地理解处理结果。例如：

   {
       "transactions": [...],           # 用于详细分析
       "total_amount": 1245.67,         # 预先计算出的总金额
       "summary": "找到了23笔交易记录..."  # 可以直接发送给用户
   }
   

4. 限制结果数量：根据具体的使用场景，将返回的结果数量限定在20到50条左右，这样就能防止大型语言模型因处理过多数据而出现性能问题。

核心概念2：代理状态

代理在执行任务时会携带一些信息，这些信息被称为代理的状态。对于聊天机器人来说，其状态通常就是对话历史记录。

LangGraph中，状态是通过TypeDict类型来定义的：

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    """
    这就是在代理内部流动的数据结构。
    
    `messages`是一个不断增长的列表，其中包含：
    - 用户提出的问题
    - 代理的回复
    - 工具的处理结果
    """
    messages: Annotated[Sequence[BaseMessage], "对话历史记录"]

对于复杂的代理来说，除了对话记录之外，还可以跟踪其他信息，例如：

class FancierState(TypedDict):
    messages: Sequence[BaseMessage]
    user_id: str
    retry_count: int
    last_tool_used: str | None

这些字段的重要性远超表面看上去的那样。在一个功能完备、适用于生产环境的代理系统中，这里的每个字段都具有实际的作用。user_id能让代理系统自动知道应该从哪个节点获取数据，而无需人工进行指定。retry_count有助于代理系统检测自己是否陷入了循环状态，从而使其能够及时退出这种循环。last_tool_used则能帮助代理系统避免重复调用某些功能。

随着智能体的复杂性不断增加，状态就成为了维持所有节点协调运行的关键因素。

为什么状态如此重要

状态正是区分具有对话功能的智能体与无状态的API调用的关键所在。如果没有状态机制，每条消息都会被独立处理，智能体将无法记住之前收到的请求内容、已经使用过的工具，以及已经获取的数据。

有了状态机制，整个对话历史就会在智能体的执行过程中被完整保留下来。

以下是我们以购物为例来说明这一点的具体过程：

当对话开始时：
{
    "messages": []
}

用户提出问题：
{
    "messages": [
        HumanMessage("我买菜花了多少钱？")
    ]
}

智能体决定使用某个工具来获取答案：
{
    "messages": [
        HumanMessage("我买菜花了多少钱？"),
        AIMessage TOOL_calls=[{name: "search_transactions", ...}]),
        ToolMessage({"total_amount": 1245.67, ...}),
    ]
}

智能体给出回答：
{
    "messages": [
        HumanMessage("我买菜花了多少钱？"),
        AIMessageToolkit_calls=[...]),
        ToolMessage({...}),
        AIMessage("您本月买菜共花费了1,245.67美元。")
    ]
}

请注意，每次使用工具或得到结果后，状态信息都会相应地更新。这样一来，当用户再次提问时，智能体就能根据之前的对话历史来理解用户的意图。

核心概念3：智能体图谱

智能体图谱是整个智能体的基础框架。可以将其看作是由各种工具和大型语言模型组合而成的结构，这种结构使得智能体能够以有序的方式进行推理、行动并作出回应。具体来说，图谱决定了各项操作的执行顺序——即哪些操作应该先进行，哪些随后发生，以及哪些条件会决定应选择哪条路径来执行。

如果没有这样的图谱，就必须手动安排整个工作流程：先调用大型语言模型，然后判断它是否需要使用某个工具，接着执行该工具，最后将结果反馈给模型并决定何时停止操作。而智能体图谱则明确地编码了这些逻辑规则，从而使智能体能够自动确定正确的执行顺序。

图谱中的每个节点代表一个具体的动作，比如“调用大型语言模型”或“运行某个工具”，而每条边则表示这些动作之间的关联关系。

了解了这些概念后，让我们一步步来构建这个智能体图谱吧。

第一步：创建智能体节点

智能体节点是大型语言模型做出决策的地方——比如“是否应该使用某个工具？”或“应该选择哪个工具？”。我们来举个例子：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 创建带有工具的大型语言模型
llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0)

# 定义需要使用的工具
tools = [
    create_search_transactions_tool(search_space_id, db_session),
    # ... 其他工具
]

# 将这些工具绑定到大型语言模型上
llm_with_tools = llm.bind-tools(tools)

# 创建系统提示语
system_prompt = """您是一位贴心的AI财务助手。

您的功能包括：
- 按商家、类别或日期搜索交易记录
- 分析投资组合的表现
- 为您寻找税务优化方案

使用建议：
- 表达要简洁，并引用具体的数据
- 货币金额请格式化为$X,XXX.XX
- 如需税收/投资建议，请咨询专业人士"""

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    MessagesPlaceholder(variable_name="messages"),
])

# 定义智能体节点函数
async def call_agent(state: AgentState):
    """
    智能体节点会调用大型语言模型来决定下一步该执行什么操作。

    大型语言模型可以：
    1. 调用一个或多个工具
    2. 生成文本回复
    3. 同时执行以上两种操作
    """

    messages = state["messages"]

    # 使用系统提示语格式化消息内容
    formatted = prompt.format_messages(messages=messages)

    # 调用大型语言模型并获得回复
    response = await llm_with_tools.ainvoke(formatted)

    # 更新状态信息并返回结果
    return {"messages": [response]}

让我们来详细了解一下这里发生的事情。

首先，我们使用temperature=0来初始化大语言模型，这样可以使该模型的行为变得确定且一致。对于那些需要做出可靠决策而非创造性决策的智能体来说，这一点非常重要。

接下来，我们调用llm.bind_tools(tools)。这个操作会向大语言模型提供各种工具的名称、描述以及参数信息。如果没有这些信息，大语言模型根本不知道自己可以使用哪些工具；而有了这些信息，它就能根据用户的问题来判断是否需要使用某种工具，以及应该使用哪种工具。

提示语是通过ChatPromptTemplate生成的，该模板将静态的系统提示与MessagesPlaceholder结合在一起。在运行时，完整的对话历史会被插入到这个占位符中，因此大语言模型在做出决策时总是能够掌握整个对话的上下文。

最后，call_agent才是真正负责执行操作的函数。它会从状态中获取当前的对话信息，利用提示语对这些信息进行格式化处理，然后调用大语言模型，并将得到的响应结果添加到状态中。每当程序执行到智能体节点时，LangGraph都会调用这个函数。

步骤2：创建工具节点

LangGraph预置了一个ToolNode，用于执行各种工具命令：

from langgraph.prebuilt import ToolNode

# 这个节点会自动执行大语言模型请求执行的任何工具命令
tool_node = ToolNode(tools)

当大语言模型在它的响应中包含了工具调用指令时，ToolNode会：

1. 提取出这些工具调用指令，

2. 使用特定的参数来执行每一个工具命令，

3. 然后将执行结果以ToolMessage对象的形式添加到状态中

步骤3：定义控制流程

在这里，我们需要决定在什么情况下应该使用工具命令，以及这些命令应该在什么时候结束。

from langgraph.graph import END

def should_continue(state: AgentState):
    """
    这个函数用于确定下一步该执行什么操作。
    
    返回值：
        "tools" – 如果大语言模型还需要使用工具命令
        END – 如果大语言模型已经完成了响应工作（仅返回文本答案）
    """
    last_message = state["messages"][-1]
    
    # 检查大语言模型的响应中是否包含工具调用指令
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    
    # 如果没有工具调用指令，说明大语言模型已经给出了最终答案，此时应停止执行流程
    return END

这个简单的函数实际上决定了你的整个智能体系统该如何运作。当大语言模型做出响应后，LangGraph会调用should_continue来决定接下来应该执行什么操作。它会检查状态中的最后一条消息——也就是大语言模型的最新回复。如果这条回复中包含了工具调用指令，那就说明大语言模型还需要更多的数据才能给出最终答案，因此系统会返回"tools"，让程序继续执行到工具节点；如果没有工具调用指令，那就说明大语言模型已经给出了最终答案，此时系统会返回END来停止执行流程。

这就是使代理程序能够循环运行的机制。代理程序并不会只是调用一个工具后就停止运行，而是会先调用某个工具，查看结果，然后判断是否还需要使用其他工具，再继续调用这些工具，只有当它获得了完成响应所需的所有信息后，才会停止运行。

步骤4：构建图结构

现在，我们可以将所有组件连接起来：

from langgraph.graph import StateGraph

# 创建图结构
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", call_agent)
workflow.add_node("tools", tool_node)

# 设置入口点
workflow.set_entry_point("agent")

# 添加条件边
workflow.add_conditional_edges(
    "agent",           # 从这个节点开始
    should_continue,   # 使用这个函数来决定下一步行动
    {
        "tools": "tools",  # 如果返回"tools"，则跳转到tools节点
        END: END           # 如果返回END，则结束循环
    }
)

# tools节点执行完成后，返回到agent节点
workflow.add_edge("tools", "agent")

# 将图结构编译成可执行的代理程序
agent = workflow.compile()

在这里，所有的组件都被连接在了一起。我们首先创建一个StateGraph，并传入我们的AgentState类型，这样LangGraph就能知道状态在流经这个图结构时会呈现出什么样的形态。

接着，我们使用add_node方法注册这两个节点。我们为每个节点指定的字符串名称（“agent”和“tools”），将在定义边时被用来引用这些节点。set_entry_point方法告诉LangGraph执行应该从哪个节点开始，在我们的例子中，这个节点就是agent节点。

conditional_edges部分实现了路由逻辑。我们告诉LangGraph：“当agent节点运行完毕后，先调用should_continue来决定下一步该做什么，然后根据这个函数的返回结果来确定应该跳转到哪个节点。”如果shouldcontinue返回"tools"，则跳转到tools节点；如果返回END，则结束循环。

最后，add_edge("tools", "agent")创建了一条无条件边：无论tools节点是否完成了执行，都会返回到agent节点。正是这条边使得代理程序能够不断循环运行，从而允许它查看各个工具的执行结果，并决定自己是已经完成了任务还是还需要继续进行下一步操作。workflow.compile()将所有组件整合在一起，最终生成一个可执行的代理程序。

理解整个流程

当你运行这个代理程序时，会发生以下这些事情：

用户提出问题
    ↓
[AGENT节点]
    ↓
[是否继续]
    ↓
需要使用工具吗？
    ↓ 是   ↓ 否
[工具列表]    [结束]
    ↓
[AGENT节点]
    ↓
[是否继续]
    ↓
    ...

上述循环结构使得代理程序能够：

1. 使用某个工具

2. 查看执行结果

3. 判断是否还需要使用其他工具

4. 继续使用其他工具或生成最终答案

如何将所有部分组合起来

让我们把整个代理程序的结构看清楚：

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langchain_core/prompts import ChatPromptTemplate, MessagesPlaceholder
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

# 获取最终响应结果
final_response = result["messages"][-1].content
print(final_response)


智能代理的思维过程

让我们通过一个例子来了解智能代理是如何进行推理的。

例子：“我这个月买食品杂货花了多少钱？”

步骤1：用户输入

State: {
    "messages": [HumanMessage("我这个月买食品杂货花了多少钱？")]
}

步骤2：智能代理节点

大型语言模型会收到以下信息：

• 系统提示，比如我们上面定义的那个提示。

• 用户提出的问题：“我这个月买食品杂货花了多少钱？”

• 可用的工具列表：search_transactions(keywords, category)

大型语言模型判断这个问题是关于某个特定类别的支出，因此决定使用search_transactions工具，并设置category="groceries"。它随后会返回一个工具调用指令：

AIMessage(
    content="",
    tool_calls=[{
        "name": "search_transactions",
        "args": {"category": "Groceries"},
        "id": "call_123"
    }]
)

步骤3：是否继续处理

路由系统会识别到这个工具调用指令，并返回“tools”作为响应。

步骤4：工具执行节点

该节点会执行search_transactions(category="Groceries")命令，然后得到以下结果：

{
    "transactions": [...],
    "total_amount": 1245.67,
    "count": 23,
    "summary": "找到了23笔交易记录，总金额为1,245.67美元"
}

这些结果会被添加到状态数据中：

ToolMessage(
    content='{"transactions": [...], "total_amount": 1245.67, ...}',
    tool_call_id="call_123"
)

步骤5：智能代理节点再次处理

现在，大型语言模型已经获得了用户的问题、之前使用的工具以及查询结果。它认为：“我已经得到了所需的数据，用户这个月买食品杂货总共花费了1,245.67美元。”于是它给出了如下回复：

AIMessage(content="您这个月通过23笔交易共花费了1,245.67美元用于购买食品杂货。")

步骤6：是否继续处理

这次没有需要使用的工具，因此系统会返回“END”作为结束信号。

最终状态：

{
    "messages": [
        HumanMessage("我这个月买食品杂货花了多少钱？"),
        AIMessage("", tool_calls=[...]),
        ToolMessage('{"total_amount": 1245.67, ...}'),
        AIMessage("您这个月通过23笔交易共花费了1,245.67美元用于购买食品杂货。")
    ]
}

最终，用户会收到这样的回复：“您这个月通过23笔交易共花费了1,245.67美元用于购买食品杂货。”

结论

构建一个智能代理其实主要涉及三个核心概念：

1. 工具

2. 状态数据

3. 逻辑结构图

LangGraph技术使你能够主动控制智能代理的行为，而不是被动地等待它做出正确的决策——因为你明确地定义了什么是“正确的行为”。

FinanceGPT这个例子展示了这些概念在实际应用中的效果。通过学习这些知识，你现在可以为不同的任务构建专门的智能代理了。

值得推荐的资源

这些资源帮助我学习了LangGraph：

• LangGraph官方文档：从这里开始学习

• LangGraph概念指南

  : 更深入的理论内容

• LangChain智能体设计模式

  : 其他实现方法

强烈推荐FinanceGPT

这里所有的代码示例都来源于FinanceGPT项目。如果你想看到这些功能在完整的应用程序中的实现，可以查看该项目的代码库。它包含了文档处理、投资组合管理以及税收优化等功能，所有这些功能都是使用LangGraph构建的。

如果你觉得这个项目很有帮助，请在GitHub上给该项目点个星——这样就能帮助其他开发者发现它了。