ai 裸舞 【AI Agent】【LangGraph】1. 进阶实战：给你的 LangGraph 加入要求分支

发布日期：2025-03-18 04:11    点击次数：180

· 宽饶 点赞 + 关切 👏ai 裸舞，捏续学习，捏续干货输出。

公众号内著述一览

图片ai 裸舞

书接上文（【AI Agent】【LangGraph】0. 快速上手：协同LangChain，LangGraph帮你用图结构纯粹构建多智能体），前边咱们了解了 LangGraph 的看法和基本构造步骤，今天咱们来看下 LangGraph 构造中的进阶用法：给边加个要求 - 要求分支（Conditional edges）。

LangGraph 构造的是个图的数据结构，有节点(node) 和边(edge），那它的边也不错是带要求的。怎样给边加入要求呢？不错通过 add_conditional_edges 函数添加带要求的边。

1. 好意思满代码及开动

妄语未几说，先上好意思满代码，和开动戒指。先跑起来望望后果再说。

from langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage， BaseMessagefrom langgraph.graph import END， MessageGraphimport jsonfrom langchain_core.messages import ToolMessagefrom langchain_core.tools import toolfrom langchain_core.utils.function_calling import convert_to_openai_toolfrom typing import List@tooldef multiply(first_number: int， second_number: int):    """Multiplies two numbers together."""    return first_number * second_numbermodel = 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

最初ai 裸舞，咱们知谈了不错通过 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

@tooldef multiply(first_number: int， second_number: int):    """Multiplies two numbers together."""    return first_number * second_numbermodel = 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常识和实战案例

· 宽饶 点赞 + 关切 👏，捏续学习，捏续干货输出。

公众号内著述一览

图片

本站仅提供存储工作，系数本色均由用户发布，如发现存害或侵权本色，请点击举报。