Type something to search...
to navigateto selectESC to close
从零到英雄：使用 LangGraph 快速构建智能聊天机器人

从零到英雄：使用 LangGraph 快速构建智能聊天机器人

在这个全面的快速入门指南中，我们将使用 LangGraph 构建一个支持聊天机器人，它可以：

  • 通过搜索网络回答常见问题
  • 在调用之间保持对话状态
  • 将复杂查询路由到人工进行审查
  • 使用自定义状态来控制其行为
  • 回溯并探索替代对话路径

我们将从一个基本的聊天机器人开始，逐步添加更复杂的功能，同时介绍关键的 LangGraph 概念。

设置

首先，安装所需的包：

%pip install -U langgraph langsmith langchain_anthropic

接下来，设置您的 API 密钥：

import getpass
import os

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("ANTHROPIC_API_KEY")

为 LangGraph 开发设置 LangSmith。注册 LangSmith 以快速发现问题并提高您的 LangGraph 项目的性能。LangSmith 允许您使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序。

第1部分：构建一个基本的聊天机器人

我们将首先使用 LangGraph 创建一个简单的聊天机器人。这个聊天机器人将直接对用户消息做出回应。尽管简单，它将阐明使用 LangGraph 构建的核心概念。在本节结束时，您将构建一个初步的聊天机器人。

定义状态图

首先创建一个 StateGraph。StateGraph 对象定义了我们聊天机器人的结构，作为一个 状态机。我们将添加节点来表示 LLM 和我们的聊天机器人可以调用的函数，并添加边来指定机器人如何在这些函数之间转换。

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


class State(TypedDict):
    # 消息的类型是 "list"。`add_messages` 函数
    # 在注释中定义了如何更新此状态键
    # （在这种情况下，它将消息附加到列表中，而不是覆盖它们）
    messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)

注意：

  • 状态定义：State 包含图的架构以及指定如何将更新应用于状态的 reducer 函数。
  • 注释消息：messages 键被注释为 add_messages reducer 函数，这告诉 LangGraph 将新消息附加到现有列表中，而不是覆盖它。

添加聊天机器人节点

接下来，添加一个 chatbot 节点。节点代表工作单元，通常是常规的 Python 函数。

from langchain_anthropic import ChatAnthropic
llm = ChatAnthropic(model="claude-2")

def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}
## The first argument is the unique node name
## The second argument is the function or object that will be called whenever
## the node is used.
graph_builder.add_node("chatbot", chatbot)
  • 函数说明：chatbot 函数以当前 State 作为输入，并返回一个字典，字典中包含一个更新后的 messages 列表，键为 "messages"。

定义入口和出口

为图形添加一个入口点和一个结束点。

graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
  • 入口点：告诉图形每次运行时从哪里开始工作。
  • 结束点：指示图形在 chatbot 节点之后退出。

编译图形

编译图形以创建一个 CompiledGraph，我们可以在我们的状态上调用它。

graph = graph_builder.compile()

可视化图形（可选）

您可以使用 get_graph 方法和其中一种绘图方法来可视化图形。

from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    pass  # Visualization is optional

输出:

运行聊天机器人

现在让我们运行聊天机器人！

def stream_graph_updates(user_input: str):
    for event in graph.stream({"messages": [("user", user_input)]}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

while True:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except:
        # Fallback if input() is not available
        user_input = "What do you know about LangGraph?"
        print("User: " + user_input)
        stream_graph_updates(user_input)
        break

示例互动：

User: What is LangGraph?
Assistant: LangGraph is a library designed to help build stateful multi-agent applications using language models. It provides tools for creating workflows and state machines to coordinate multiple AI agents or language model interactions.
Goodbye!

恭喜！ 你已经使用 LangGraph 构建了第一个聊天机器人。这个机器人可以通过获取用户输入并使用 LLM 生成响应来进行基本对话。

Part 2: 使用工具增强聊天机器人

为了处理我们的聊天机器人无法通过记忆回答的查询，我们将集成一个网络搜索工具。我们的机器人可以使用此工具查找相关信息并提供更好的回应。

需求

在开始之前，请确保您已安装必要的软件包并设置了 API 密钥。

安装 Tavily 搜索引擎：

%pip install -U tavily-python langchain_community

设置您的 TAVILY_API_KEY：

_set_env("TAVILY_API_KEY")

定义工具

from langchain_community.tools.tavily_search import TavilySearchResults

tool = TavilySearchResults(max_results=2)
tools = [tool]

修改图形

我们将开始定义我们的图形。以下内容与第一部分类似，但我们在我们的 LLM 中添加了 bind_tools。

from typing import Annotated
from langchain_anthropic import ChatAnthropic
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


class State(TypedDict):
    messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
llm = ChatAnthropic(model="claude-2")
llm_with_tools = llm.bind_tools(tools)
def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)

添加工具节点

创建一个函数来运行被调用的工具。

import json
from langchain_core.messages import ToolMessage

class BasicToolNode:
    """一个运行在最后一个AIMessage中请求的工具的节点。"""
    def __init__(self, tools: list) -> None:
        self.tools_by_name = {tool.name: tool for tool in tools}
    def __call__(self, inputs: dict):
        if messages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raise ValueError("输入中未找到消息")
        outputs = []
        for tool_call in message.tool_calls:
            tool_result = self.tools_by_name[tool_call["name"]].invoke(
                tool_call["args"]
            )
            outputs.append(
                ToolMessage(
                    content=json.dumps(tool_result),
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"],
                )
            )
        return {"messages": outputs}
tool_node = BasicToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

定义条件边

边缘将控制流从一个节点路由到下一个节点。条件边通常包含“if”语句，以根据当前图状态路由到不同的节点。

from typing import Literal

def route_tools(state: State):
    if isinstance(state, list):
        ai_message = state[-1]
    elif messages := state.get("messages", []):
        ai_message = messages[-1]
    else:
        raise ValueError(f"No messages found in input state to tool_edge: {state}")
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"
    return END
graph_builder.add_conditional_edges(
    "chatbot",
    route_tools,
    {"tools": "tools", END: END},
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

可视化增强图

from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    pass

输出:

运行增强型聊天机器人

while True:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except:
        user_input = "What do you know about LangGraph?"
        print("User: " + user_input)
        stream_graph_updates(user_input)
        break

示例互动:

User: What's a 'node' in LangGraph?
Assistant: Based on the search results, a 'node' in LangGraph represents a function or computation step. Each node performs a specific task and can be connected to other nodes to form a workflow.
Goodbye!

恭喜！ 您已经在 LangGraph 中创建了一个可以在需要时使用搜索引擎检索更新信息的对话代理。

第3部分：为聊天机器人添加记忆

我们的聊天机器人现在可以使用工具来回答用户的问题，但它并不记得之前交互的上下文。这限制了它进行连贯的多轮对话的能力。

使用检查点技术进行内存管理

LangGraph通过持久性检查点解决了这个问题。如果在编译图时提供一个检查点器，并在调用图时提供一个thread_id，LangGraph会在每一步后自动保存状态。

创建一个 MemorySaver 检查点器：

from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()

更新图表

我们将使用 LangGraph 的预构建 ToolNode 和 tools_condition 来简化操作。

from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

class State(TypedDict):
    messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
tool = TavilySearchResults(max_results=2)
tools = [tool]
llm = ChatAnthropic(model="claude-2")
llm_with_tools = llm.bind_tools(tools)
def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)
tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile(checkpointer=memory)

与具有记忆的聊天机器人互动

设置线程 ID：

config = {"configurable": {"thread_id": "1"}}

第一次互动：

user_input = "Hi there! My name is Alice."
events = graph.stream(
    {"messages": [("user", user_input)]}, config, stream_mode="values"
)
for event in events:
    event["messages"][-1].pretty_print()

第二次互动：

user_input = "Remember my name?"
events = graph.stream(
    {"messages": [("user", user_input)]}, config, stream_mode="values"
)
for event in events:
    event["messages"][-1].pretty_print()

示例输出：

mathematica

Assistant: Of course, I remember your name, Alice. How can I assist you today?

恭喜！ 由于 LangGraph 的检查点系统，您的聊天机器人现在可以在会话之间保持对话状态。

第4部分：人机协作

代理可能不可靠，可能需要人类的输入才能成功完成任务。同样，对于某些操作，您可能希望在运行之前要求人类批准，以确保一切按预期进行。

添加人工监督

我们将使用 LangGraph 的 interrupt_before 功能来始终中断工具节点。

编译带中断的图：

graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["tools"],
)

与人类参与者的交互

用户输入：

user_input = "I'm learning LangGraph. Could you do some research on it for me?"
config = {"configurable": {"thread_id": "1"}}
events = graph.stream(
    {"messages": [("user", user_input)]}, config, stream_mode="values"
)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

检查状态：

snapshot = graph.get_state(config)
existing_message = snapshot.values["messages"][-1]
existing_message.tool_calls

继续图形：

events = graph.stream(None, config, stream_mode="values")
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

恭喜！ 您已使用中断为您的聊天机器人添加了人类参与者执行，使其在需要时能够进行人工监督和干预。

第5部分：手动更新状态

LangGraph 允许您手动更新状态，使您能够通过修改代理的行为来控制其轨迹。

更新状态

开始一个新线程：

user_input = "I'm learning LangGraph. Could you do some research on it for me?"
config = {"configurable": {"thread_id": "2"}}
events = graph.stream(
    {"messages": [("user", user_input)]}, config, stream_mode="values"
)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

修改工具调用：

from langchain_core.messages import AIMessage

snapshot = graph.get_state(config)
existing_message = snapshot.values["messages"][-1]
new_tool_call = existing_message.tool_calls[0].copy()
new_tool_call["args"]["query"] = "LangGraph human-in-the-loop workflow"
new_message = AIMessage(
    content=existing_message.content,
    tool_calls=[new_tool_call],
    id=existing_message.id,
)
graph.update_state(config, {"messages": [new_message]})

恢复图形：

events = graph.stream(None, config, stream_mode="values")
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

恭喜！ 你已经使用 interrupt_before 和 update_state 手动修改状态，作为人机协作工作流的一部分。

第6部分：自定义状态

我们将通过添加一个新节点来扩展我们的聊天机器人，以说明如何使用自定义状态更新来定制机器人的行为。

定义扩展状态

from typing import Annotated
from langchain_anthropic import ChatAnthropic
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]
    ask_human: bool  # 这个标志是新的

定义人类辅助模式

from pydantic import BaseModel

class RequestAssistance(BaseModel):
    """Escalate the conversation to an expert."""
    request: str

更新聊天机器人节点

tool = TavilySearchResults(max_results=2)
tools = [tool]
llm = ChatAnthropic(model="claude-2")
llm_with_tools = llm.bind_tools(tools + [RequestAssistance])

def chatbot(state: State):
    response = llm_with_tools.invoke(state["messages"])
    ask_human = False
    if (
        response.tool_calls
        and response.tool_calls[0]["name"] == RequestAssistance.__name__
    ):
        ask_human = True
    return {"messages": [response], "ask_human": ask_human}

添加人类节点

from langchain_core.messages import AIMessage, ToolMessage

def create_response(response: str, ai_message: AIMessage):
    return ToolMessage(
        content=response,
        tool_call_id=ai_message.tool_calls[0]["id"],
    )
def human_node(state: State):
    new_messages = []
    if not isinstance(state["messages"][-1], ToolMessage):
        new_messages.append(
            create_response("No response from human.", state["messages"][-1])
        )
    return {
        "messages": new_messages,
        "ask_human": False,
    }
graph_builder.add_node("human", human_node)

定义条件逻辑

def select_next_node(state: State):
    if state["ask_human"]:
        return "human"
    return tools_condition(state)


graph_builder.add_conditional_edges(
    "chatbot",
    select_next_node,
    {"human": "human", "tools": "tools", END: END},
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge("human", "chatbot")
graph_builder.add_edge(START, "chatbot")
memory = MemorySaver()
graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["human"],
)

与定制聊天机器人互动

用户输入：

user_input = "I need some expert guidance for building this AI agent. Could you request assistance for me?"
config = {"configurable": {"thread_id": "1"}}
events = graph.stream(
    {"messages": [("user", user_input)]}, config, stream_mode="values"
)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

作为人类的回应：

ai_message = snapshot.values["messages"][-1]
human_response = (
    "We, the experts, recommend you check out LangGraph to build your agent."
)
tool_message = create_response(human_response, ai_message)
graph.update_state(config, {"messages": [tool_message]})

恢复图形：

events = graph.stream(None, config, stream_mode="values")
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

恭喜！ 您已向助手图添加了一个额外节点，让聊天机器人自行决定是否需要中断执行。

结论

在本综合指南中，我们使用 LangGraph 构建了一个具有逐步增强功能的支持聊天机器人：

  • 基本聊天机器人：直接响应用户消息。
  • 增强工具：使用网络搜索工具回答问题。
  • 增加记忆：在调用之间维护对话状态。
  • 人类参与：允许人类监督和干预。
  • 自定义状态：通过自定义状态更新控制行为。

LangGraph 的灵活性和强大功能使开发人员能够轻松创建复杂的、有状态的 AI 应用程序。通过掌握这些概念，您在构建高级对话代理和探索 AI 开发的广阔可能性方面已经走上了良好的道路。

祝编码愉快！

参考文献：

介绍 | 🦜️🔗 LangChain

https://python.langchain.com/v0.2/docs/how_to/migrate_agent/

https://langchain-ai.github.io/langgraph/tutorials/introduction

Tags :
Share :

Related Posts

使用 ChatGPT 搜索网络功能的 10 种创意方法

使用 ChatGPT 搜索网络功能的 10 种创意方法

例如，提示和输出 你知道可以使用 ChatGPT 的“搜索网络”功能来完成许多任务，而不仅仅是基本的网络搜索吗？ 对于那些不知道的人，ChatGPT 新的“搜索网络”功能提供实时信息。 截至撰写此帖时，该功能仅对使用 ChatGPT 4o 和 4o-mini 的付费会员开放。 ![](https://images.weserv.nl/?url=https://cdn-im

阅读更多
在人工智能和技术领域保持领先地位的 10 项必学技能 📚

在人工智能和技术领域保持领先地位的 10 项必学技能 📚

在人工智能和科技这样一个动态的行业中，保持领先意味着不断提升你的技能。无论你是希望深入了解人工智能模型性能、掌握数据分析，还是希望通过人工智能转变传统领域如法律，这些课程都是你成功的捷径。以下是一个精心策划的高价值课程列表，可以助力你的职业发展，并让你始终处于创新的前沿。 1. 生成性人工智能简介课程: [生成性人工智能简介](https://genai.works

阅读更多
10 个强大的 Perplexity AI 提示，让您的营销任务自动化

10 个强大的 Perplexity AI 提示，让您的营销任务自动化

在当今快速变化的数字世界中，营销人员总是在寻找更智能的方法来简化他们的工作。想象一下，有一个个人助理可以为您创建受众档案，建议营销策略，甚至为您撰写广告文案。这听起来像是一个梦想？ 多亏了像 Perplexity 这样的 AI 工具，这个梦想现在成为现实。通过正确的提示，您可以将 AI 转变为您的 个人营销助理。在本文中，我将分享 10 个强大的提示，帮助您自动

阅读更多
10+ 面向 UI/UX 设计师的顶级 ChatGPT 提示

10+ 面向 UI/UX 设计师的顶级 ChatGPT 提示

人工智能技术，如机器学习、自然语言处理和数据分析，正在重新定义传统设计方法。从自动化重复任务到实现个性化用户体验，人工智能使设计师能够更加专注于战略思维和创造力。随着这一趋势的不断增长，UI/UX 设计师越来越多地采用 AI 驱动的工具来促进他们的工作。利用人工智能不仅能提供基于数据的洞察，还为满足多样化用户需求的创新设计解决方案开辟了机会。 1. 用户角色开发 目的

阅读更多
在几分钟内完成数月工作的 100 种人工智能工具

在几分钟内完成数月工作的 100 种人工智能工具

人工智能（AI）的快速发展改变了企业的运作方式，使人们能够在短短几分钟内完成曾经需要几周或几个月的任务。从内容创作到网站设计，AI工具帮助专业人士节省时间，提高生产力，专注于创造力。以下是按功能分类的100个AI工具的全面列表，以及它们在现实世界中的使用实例。 1. 研究工具 研究可能耗时，但人工智能工具使查找、分析和组织数据变得更加容易。**ChatGPT, Cop

阅读更多
你从未知道的 17 个令人惊叹的 GitHub 仓库

你从未知道的 17 个令人惊叹的 GitHub 仓库

Github 隐藏的宝石!! 立即收藏的代码库 学习编程相对简单，但掌握编写更好代码的艺术要困难得多。GitHub 是开发者的宝藏，那里“金子”是其他人分享的精心编写的代码。通过探索 GitHub，您可以发现如何编写更清晰的代码，理解高质量代码的样子，并学习成为更熟练开发者的基本步骤。 1. notwaldorf/emoji-translate *谁需

阅读更多