【AI Agent系列】【LangGraph】1. 进阶实战:给你的 LangGraph 加入条件分支(Conditional edges)

本文涉及的产品
阿里云百炼推荐规格 ADB PostgreSQL,4核16GB 100GB 1个月
简介: 【AI Agent系列】【LangGraph】1. 进阶实战:给你的 LangGraph 加入条件分支(Conditional edges)
  • 大家好,我是同学小张,日常分享AI知识和实战案例
  • 欢迎 点赞 + 关注 👏,持续学习持续干货输出
  • 一起交流💬,一起进步💪。
  • 微信公众号也可搜【同学小张】 🙏

本站文章一览:


书接上文(【AI Agent系列】【LangGraph】0. 快速上手:协同LangChain,LangGraph帮你用图结构轻松构建多智能体应用),前面我们了解了 LangGraph 的概念和基本构造方法,今天我们来看下 LangGraph 构造中的进阶用法:给边加个条件 - 条件分支(Conditional edges)。

LangGraph 构造的是个图的数据结构,有节点(node) 和边(edge),那它的边也可以是带条件的。如何给边加入条件呢?可以通过 add_conditional_edges 函数添加带条件的边。

1. 完整代码及运行

废话不多说,先上完整代码,和运行结果。先跑起来看看效果再说。

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, BaseMessage
from langgraph.graph import END, MessageGraph
import json
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langchain_core.utils.function_calling import convert_to_openai_tool
from typing import List
@tool
def multiply(first_number: int, second_number: int):
    """Multiplies two numbers together."""
    return first_number * second_number
model = ChatOpenAI(temperature=0)
model_with_tools = model.bind(tools=[convert_to_openai_tool(multiply)])
graph = MessageGraph()
def invoke_model(state: List[BaseMessage]):
    return model_with_tools.invoke(state)
graph.add_node("oracle", invoke_model)
def invoke_tool(state: List[BaseMessage]):
    tool_calls = state[-1].additional_kwargs.get("tool_calls", [])
    multiply_call = None
    for tool_call in tool_calls:
        if tool_call.get("function").get("name") == "multiply":
            multiply_call = tool_call
    if multiply_call is None:
        raise Exception("No adder input found.")
    res = multiply.invoke(
        json.loads(multiply_call.get("function").get("arguments"))
    )
    return ToolMessage(
        tool_call_id=multiply_call.get("id"),
        content=res
    )
graph.add_node("multiply", invoke_tool)
graph.add_edge("multiply", END)
graph.set_entry_point("oracle")
def router(state: List[BaseMessage]):
    tool_calls = state[-1].additional_kwargs.get("tool_calls", [])
    if len(tool_calls):
        return "multiply"
    else:
        return "end"
graph.add_conditional_edges("oracle", router, {
    "multiply": "multiply",
    "end": END,
})
runnable = graph.compile()
response = runnable.invoke(HumanMessage("What is 123 * 456?"))
print(response)

运行结果如下:

2. 代码详解

下面对上面的代码进行详细解释。

2.1 add_conditional_edges

首先,我们知道了可以通过 add_conditional_edges 来对边进行条件添加。这部分代码如下:

graph.add_conditional_edges("oracle", router, {
    "multiply": "multiply",
    "end": END,
})

add_conditional_edges接收三个参数:

  • 第一个为这条边的第一个node的名称
  • 第二个为这条边的条件
  • 第三个为条件返回结果的映射(根据条件结果映射到相应的node)

如上面的代码,意思就是往 “oracle” node上添加边,这个node有两条边,一条是往“multiply” node上走,一条是往“END”上走。怎么决定往哪个方向去:条件是 router(后面解释),如果 router 返回的是“multiply”,则往“multiply”方向走,如果 router 返回的是 “end”,则走“END”。

来看下这个函数的源码:

def add_conditional_edges(
    self,
    start_key: str,
    condition: Callable[..., str],
    conditional_edge_mapping: Optional[Dict[str, str]] = None,
) -> None:
    if self.compiled:
        logger.warning(
            "Adding an edge to a graph that has already been compiled. This will "
            "not be reflected in the compiled graph."
        )
    if start_key not in self.nodes:
        raise ValueError(f"Need to add_node `{start_key}` first")
    if iscoroutinefunction(condition):
        raise ValueError("Condition cannot be a coroutine function")
    if conditional_edge_mapping and set(
        conditional_edge_mapping.values()
    ).difference([END]).difference(self.nodes):
        raise ValueError(
            f"Missing nodes which are in conditional edge mapping. Mapping "
            f"contains possible destinations: "
            f"{list(conditional_edge_mapping.values())}. Possible nodes are "
            f"{list(self.nodes.keys())}."
        )
    self.branches[start_key].append(Branch(condition, conditional_edge_mapping))

重点是这一句:self.branches[start_key].append(Branch(condition, conditional_edge_mapping)),给当前node添加分支Branch。

2.2 条件 router

条件代码如下:判断执行结果中是否有 tool_calls 参数,如果有,则返回"multiply",没有,则返回“end”。

def router(state: List[BaseMessage]):
    tool_calls = state[-1].additional_kwargs.get("tool_calls", [])
    if len(tool_calls):
        return "multiply"
    else:
        return "end"

2.3 各node的定义

(1)起始node:oracle

@tool
def multiply(first_number: int, second_number: int):
    """Multiplies two numbers together."""
    return first_number * second_number
model = ChatOpenAI(temperature=0)
model_with_tools = model.bind(tools=[convert_to_openai_tool(multiply)])
graph = MessageGraph()
def invoke_model(state: List[BaseMessage]):
    return model_with_tools.invoke(state)
graph.add_node("oracle", invoke_model)

这个node是一个带有Tools 的 ChatOpenAI。在LangChain中使用Tools的详细教程请看这篇文章:【AI大模型应用开发】【LangChain系列】5. 实战LangChain的智能体Agents模块。简单解释就是:这个node的执行结果,将返回是否应该使用绑定的Tools。

(2)multiply

def invoke_tool(state: List[BaseMessage]):
    tool_calls = state[-1].additional_kwargs.get("tool_calls", [])
    multiply_call = None
    for tool_call in tool_calls:
        if tool_call.get("function").get("name") == "multiply":
            multiply_call = tool_call
    if multiply_call is None:
        raise Exception("No adder input found.")
    res = multiply.invoke(
        json.loads(multiply_call.get("function").get("arguments"))
    )
    return ToolMessage(
        tool_call_id=multiply_call.get("id"),
        content=res
    )
graph.add_node("multiply", invoke_tool)

这个node的作用就是执行Tools。

2.4 总体流程

如果觉得本文对你有帮助,麻烦点个赞和关注呗 ~~~


  • 大家好,我是 同学小张,日常分享AI知识和实战案例
  • 欢迎 点赞 + 关注 👏,持续学习持续干货输出
  • 一起交流💬,一起进步💪。
  • 微信公众号也可搜【同学小张】 🙏

本站文章一览:

相关实践学习
AnalyticDB PostgreSQL 企业智能数据中台:一站式管理数据服务资产
企业在数据仓库之上可构建丰富的数据服务用以支持数据应用及业务场景;ADB PG推出全新企业智能数据平台,用以帮助用户一站式的管理企业数据服务资产,包括创建, 管理,探索, 监控等; 助力企业在现有平台之上快速构建起数据服务资产体系
相关文章
|
21天前
|
存储 消息中间件 人工智能
【03】AI辅助编程完整的安卓二次商业实战-本地构建运行并且调试-二次开发改注册登陆按钮颜色以及整体资源结构熟悉-优雅草伊凡
【03】AI辅助编程完整的安卓二次商业实战-本地构建运行并且调试-二次开发改注册登陆按钮颜色以及整体资源结构熟悉-优雅草伊凡
56 3
|
21天前
|
机器学习/深度学习 数据采集 人工智能
PyTorch学习实战:AI从数学基础到模型优化全流程精解
本文系统讲解人工智能、机器学习与深度学习的层级关系,涵盖PyTorch环境配置、张量操作、数据预处理、神经网络基础及模型训练全流程,结合数学原理与代码实践,深入浅出地介绍激活函数、反向传播等核心概念,助力快速入门深度学习。
74 1
|
17天前
|
人工智能 自然语言处理 API
快速集成GPT-4o:下一代多模态AI实战指南
快速集成GPT-4o:下一代多模态AI实战指南
210 101
|
21天前
|
人工智能 JSON 测试技术
AI智能体开发实战:从提示工程转向上下文工程的完整指南
曾被热捧的提示工程正逐渐退潮,本文揭示其局限性,并提出“上下文工程”新范式:通过结构化提示、精准上下文管理、工具调用与统一状态,构建可扩展、可恢复、生产级的智能体工作流,推动AI系统迈向工程化与可控化。
193 9
AI智能体开发实战:从提示工程转向上下文工程的完整指南
|
21天前
|
存储 消息中间件 人工智能
【08】AI辅助编程完整的安卓二次商业实战-修改消息聊天框背景色-触发聊天让程序异常终止bug牵涉更多聊天消息发送优化处理-优雅草卓伊凡
【08】AI辅助编程完整的安卓二次商业实战-修改消息聊天框背景色-触发聊天让程序异常终止bug牵涉更多聊天消息发送优化处理-优雅草卓伊凡
106 10
【08】AI辅助编程完整的安卓二次商业实战-修改消息聊天框背景色-触发聊天让程序异常终止bug牵涉更多聊天消息发送优化处理-优雅草卓伊凡
|
21天前
|
存储 消息中间件 人工智能
【05】AI辅助编程完整的安卓二次商业实战-消息页面媒体对象(Media Object)布局实战调整-按钮样式调整实践-优雅草伊凡
【05】AI辅助编程完整的安卓二次商业实战-消息页面媒体对象(Media Object)布局实战调整-按钮样式调整实践-优雅草伊凡
65 11
【05】AI辅助编程完整的安卓二次商业实战-消息页面媒体对象(Media Object)布局实战调整-按钮样式调整实践-优雅草伊凡
|
17天前
|
机器学习/深度学习 算法 数据可视化
从零开始训练推理模型:GRPO+Unsloth改造Qwen实战指南
推理型大语言模型兴起,通过先思考再作答提升性能。本文介绍GRPO等强化学习算法,详解其原理并动手用Qwen2.5-3B训练推理模型,展示训练前后效果对比,揭示思维链生成的实现路径。
153 1
从零开始训练推理模型:GRPO+Unsloth改造Qwen实战指南
|
21天前
|
XML 存储 Java
【06】AI辅助编程完整的安卓二次商业实战-背景布局变更增加背景-二开发现页面跳转逻辑-替换剩余图标-优雅草卓伊凡
【06】AI辅助编程完整的安卓二次商业实战-背景布局变更增加背景-二开发现页面跳转逻辑-替换剩余图标-优雅草卓伊凡
55 3
【06】AI辅助编程完整的安卓二次商业实战-背景布局变更增加背景-二开发现页面跳转逻辑-替换剩余图标-优雅草卓伊凡
|
21天前
|
存储 消息中间件 人工智能
【04】AI辅助编程完整的安卓二次商业实战-寻找修改替换新UI首页图标-菜单图标-消息列表图标-优雅草伊凡
【04】AI辅助编程完整的安卓二次商业实战-寻找修改替换新UI首页图标-菜单图标-消息列表图标-优雅草伊凡
52 4
|
21天前
|
数据采集 人工智能 JSON
学会“读网页”:生成式 AI 在足球赛事信息整理中的实战
本文介绍了一个足球比赛信息提取教程,利用生成式AI从ESPN、虎扑、腾讯体育等网站抓取比赛报道,抽取比分、关键事件和球员表现等信息。步骤包括采集、清洗、分块、调用LLM抽取、校验与落地,以及可视化。需要准备Python环境、安装依赖库,并遵循合规性。提供了示例代码,演示如何从网页抓取文本并调用LLM接口获得结构化输出。

热门文章

最新文章