欢迎光临
我们一直在努力

LangGraph 入门指南

        常见的大模型回答问题,往往是单论交互:输入问题—->模型输出结果,End。

        而智能体应用,需要一个多步骤、有状态、会决策的流程,例如:客服智能体:一般要根据用户输入的订单号,先查询订单信息,并根据具体情况转人工或者走退款审批流程。这类场景的特点是:

        1、有明确的步骤顺序:先执行A操作,再根据结果决定下一步。

        2、存在分支和循环,查不到信息可以执行重试,权限不足转人工等等。

        3、需要记忆与状态。

        LangGraph:专门为构造这种“有状态、多步骤、可控的智能体”而产生的框架。是一个用“图”的方式来定义和运行智能体工作流的库。智能体被建模成一个有节点(Node)和边(Edge)的有向图,每个节点是一个计算步骤,边则控制数据流动的下一步走向。langGraph使用的图是数学分支——图论(Graph Theory)理论体系下的产物。

核心概念(用最简单的比喻)

想象你在做一个“请假审批”的流程图:

  • 节点(Node):一个处理步骤,比如“主管审批”、“人事备案”。
    在 LangGraph 中,节点可以是调用大模型的函数、调用工具的函数、或是纯粹的逻辑判断。

  • 边(Edge):连接节点的箭头,表示“下一步去哪儿”。边可以是固定的,也可以是根据条件动态选择的。

  • 状态(State):整个流程共享的一个数据对象(通常是一个字典),就像流程图中流转的“表单”。每个节点都可以读取和更新这个状态。

  • 图(Graph):由节点和边组成的完整工作流结构,你可以把它编译成一个可运行的程序。

  • 为什么选择 LangGraph?(相比自己写循环)

  • 支持循环和条件分支:你自己写 while 循环很容易失控(无限循环、状态丢失),而 LangGraph 通过图编译,能自动管理执行流,还支持设置递归上限。

  • 持久化和断点恢复:LangGraph 能一行代码就实现“检查点”,让智能体流程在任意节点中断后,重新运行时从断点继续。这在需要“等待人类批准”的场景里非常关键。

  • 流式输出:天然支持逐步输出每个节点的结果,方便构建实时交互的应用。

  • 可视化和调试:图结构本身就是文档,而且可以用 LangSmith 等工具调试流程。

  • 一个极简示例:

    定义class:

    from typing import TypedDict

    class AgentState(TypedDict):
    messages: list # 存放历史对话列表
    ask_again: bool # 控制是否再问一次

    定义节点:

    from langchain_openai import ChatOpenAI

    model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

    # 节点1:生成初次回复
    def generate_first_reply(state: AgentState):
    response = model.invoke(state["messages"])
    # 简单的条件:如果回复里有“?”就标记再次询问
    ask = "?" in response.content
    return {"messages": [response], "ask_again": ask}

    # 节点2:在条件满足时,基于前面内容再追问一个后续问题
    def ask_followup(state: AgentState):
    # 将助手的上一条回复和“请追问一个相关问题”加入提示
    last_msg = state["messages"][-1]
    prompt = f"Based on: {last_msg.content}\\nAsk an interesting follow-up question:"
    response = model.invoke(prompt)
    return {"messages": [response], "ask_again": False}

    构建图:

    from langgraph.graph import StateGraph, END

    graph = StateGraph(AgentState)

    # 添加节点
    graph.add_node("first_reply", generate_first_reply)
    graph.add_node("followup", ask_followup)

    # 设置入口
    graph.set_entry_point("first_reply")

    # 添加条件边:根据 ask_again 决定去 followup 还是结束
    def should_continue(state: AgentState):
    if state["ask_again"]:
    return "followup"
    return END

    graph.add_conditional_edges("first_reply", should_continue)
    graph.add_edge("followup", END) # followup 节点后直接结束

    # 编译成可运行应用
    app = graph.compile()

    调用方:

    inputs = {"messages": ["Hello, how are you?"]}
    for output in app.stream(inputs):
    # 每个节点的输出都会流式打印
    print(output)

    赞(0)
    未经允许不得转载:171主机测评 » LangGraph 入门指南
    分享到: 更多 (0)

    评论 抢沙发

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址