一、LangGraph基础入门

推荐语

LangGraph框架助力构建智能协作应用，解锁多智能体系统的循环与分支控制能力。

核心内容：
1. LangGraph的核心架构与状态管理机制
2. 支持循环/分支/持久化的关键功能解析
3. 快速入门指南与工具集成实践

杨芳贤

53A创始人/腾讯云(TVP)最具价值专家

1 LangGraph介绍

LangGraph是由LangChain Inc开发的开源框架，专为构建有状态、多智能体（Multi-Agent）协作应用而设计‌。其核心通过图结构（Graph）实现复杂任务的动态编排，支持循环、分支、并行等流程控制，适用于客服系统、自动化决策、多轮对话等场景。

LangGraph 是 LangChain 的高级库，为大型语言模型（LLM）带来循环计算能力。它超越了 LangChain 的线性工作流，通过循环支持复杂的任务处理。

• 状态：维护计算过程中的上下文，实现基于累积数据的动态决策。
• 节点：代表计算步骤，执行特定任务，可定制以适应不同工作流。
• 边：连接节点，定义计算流程，支持条件逻辑，实现复杂工作流。
  
  

2 LangChain主要功能

2.1 LangChain库

• 循环和分支：在您的应用程序中实现循环和条件语句。
• 持久性：在图中的每个步骤之后自动保存状态。在任何时候暂停和恢复图执行以支持错误恢复、“人机交互”工作流、时间旅行等等。
• “人机交互”：中断图执行以批准或编辑代理计划的下一个动作。
• 流支持：在每个节点产生输出时流式传输输出（包括令牌流式传输）。
• 与 LangChain 集成：LangGraph 与LangChain和LangSmith无缝集成。

3 LangChain快速入门

3.1 安装LangChaph

要安装LangChaph，可以使用Pip和Conda进行安装。以下是安装LangGraph（国内镜像源）的步骤：

pip install -U langgraph -i https://pypi.tuna.tsinghua.edu.cn/simple

3.2 LangGraph示例

LangGraph 的一个核心概念是状态。每次图执行都会创建一个状态，该状态在图中的节点执行时传递，每个节点在执行后使用其返回值更新此内部状态。图更新其内部状态的方式由所选图类型或自定义函数定义。

让我们看一个可以使用搜索工具的简单代理示例。

#示例：langgraph_hello.py
fromtypingimportLiteral
fromlangchain_core.messagesimportHumanMessage
fromlangchain_core.toolsimporttool
fromlangchain_openaiimportChatOpenAI
# pip install langgraph
fromlanggraph.checkpoint.memoryimportMemorySaver
fromlanggraph.graphimportEND, StateGraph, MessagesState
fromlanggraph.prebuiltimportToolNode

importos
os.environ["DASHSCOPE_API_KEY"] = "替换API"


model = ChatOpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model="qwen-vl-max-latest"
)

# 定义工具函数，用于代理调用外部工具
@tool
defsearch(query: str):
    """模拟一个搜索工具"""
    if"上海"inquery.lower() or"Shanghai"inquery.lower():
        return"现在30度，有雾."
    return"现在是35度，阳光明媚。"


# 将工具函数放入工具列表
tools = [search]

# 创建工具节点
tool_node = ToolNode(tools)

# 定义函数，决定是否继续执行
defshould_continue(state: MessagesState) ->Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]
    # 如果LLM调用了工具，则转到“tools”节点
    iflast_message.tool_calls:
        return"tools"
    # 否则，停止（回复用户）
    returnEND


# 定义调用模型的函数
defcall_model(state: MessagesState):
    messages = state['messages']
    response = model.invoke(messages)
    # 返回列表，因为这将被添加到现有列表中
    return {"messages": [response]}

# 2.用状态初始化图，定义一个新的状态图
workflow = StateGraph(MessagesState)
# 3.定义图节点，定义我们将循环的两个节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# 4.定义入口点和图边
# 设置入口点为“agent”
# 这意味着这是第一个被调用的节点
workflow.set_entry_point("agent")

# 添加条件边
workflow.add_conditional_edges(
    # 首先，定义起始节点。我们使用`agent`。
    # 这意味着这些边是在调用`agent`节点后采取的。
    "agent",
    # 接下来，传递决定下一个调用节点的函数。
    should_continue,
)

# 添加从`tools`到`agent`的普通边。
# 这意味着在调用`tools`后，接下来调用`agent`节点。
workflow.add_edge("tools", 'agent')

# 初始化内存以在图运行之间持久化状态
checkpointer = MemorySaver()

# 5.编译图
# 这将其编译成一个LangChain可运行对象，
# 这意味着你可以像使用其他可运行对象一样使用它。
# 注意，我们（可选地）在编译图时传递内存
app = workflow.compile(checkpointer=checkpointer)

# 6.执行图，使用可运行对象
final_state = app.invoke(
    {"messages": [HumanMessage(content="上海的天气怎么样?")]},
    config={"configurable": {"thread_id": 42}}
)
# 从 final_state 中获取最后一条消息的内容
result = final_state["messages"][-1].content
print(result)
final_state = app.invoke(
    {"messages": [HumanMessage(content="我问的那个城市?")]},
    config={"configurable": {"thread_id": 42}}
)
result = final_state["messages"][-1].content
print(result)