链载Ai
标题: 讲解 LangGraph 构造中的进阶用法 [打印本页]
作者: 链载Ai 时间: 2 小时前
标题: 讲解 LangGraph 构造中的进阶用法
ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;margin: 1.5em 8px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">书接上文（【AI Agent】【LangGraph】0. 快速上手：协同LangChain，LangGraph帮你用图结构轻松构建多智能体），前面我们了解了 LangGraph 的概念和基本构造方法，今天我们来看下 LangGraph 构造中的进阶用法：给边加个条件 - 条件分支（Conditional edges）。ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;margin: 1.5em 8px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">LangGraph 构造的是个图的数据结构，有节点(node) 和边(edge），那它的边也可以是带条件的。如何给边加入条件呢？可以通过add_conditional_edges函数添加带条件的边。ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 1.2em;font-weight: bold;display: table;margin: 2em auto 1em;padding-right: 1em;padding-left: 1em;border-bottom: 2px solid rgb(15, 76, 129);color: rgb(63, 63, 63);">1. 完整代码及运行ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;margin: 1.5em 8px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">废话不多说，先上完整代码，和运行结果。先跑起来看看效果再说。ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;overflow-x: auto;border-radius: 8px;padding: 1em;margin: 10px 8px;">fromlangchain_openaiimportChatOpenAI
fromlangchain_core.messagesimportHumanMessage,BaseMessage
fromlanggraph.graphimportEND,MessageGraph
importjson
fromlangchain_core.messagesimportToolMessage
fromlangchain_core.toolsimporttool
fromlangchain_core.utils.function_callingimportconvert_to_openai_tool
fromtypingimportList
@tool
defmultiply(first_number:int,second_number:int):
"""Multipliestwonumberstogether."""
returnfirst_number*second_number
model=ChatOpenAI(temperature=0)
model_with_tools=model.bind(tools=[convert_to_openai_tool(multiply)])
graph=MessageGraph()
definvoke_model(state
ist[BaseMessage]):
returnmodel_with_tools.invoke(state)
graph.add_node("oracle",invoke_model)
definvoke_tool(stateist[BaseMessage]):
tool_calls=state[-1].additional_kwargs.get("tool_calls",[])
multiply_call=None
fortool_callintool_calls:
iftool_call.get("function").get("name")=="multiply":
multiply_call=tool_call
ifmultiply_callisNone:
raiseException("Noadderinputfound.")
res=multiply.invoke(
json.loads(multiply_call.get("function").get("arguments"))
)
returnToolMessage(
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")
defrouter(stateist[BaseMessage]):
tool_calls=state[-1].additional_kwargs.get("tool_calls",[])
iflen(tool_calls):
return"multiply"
else:
return"end"
graph.add_conditional_edges("oracle",router,{
"multiply":"multiply",
"end":END,
})
runnable=graph.compile()
response=runnable.invoke(HumanMessage("Whatis123*456?"))
print(response)ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;margin: 1.5em 8px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">运行结果如下：
ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 1.2em;font-weight: bold;display: table;margin: 2em auto 1em;padding-right: 1em;padding-left: 1em;border-bottom: 2px solid rgb(15, 76, 129);color: rgb(63, 63, 63);">2. 代码详解ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;margin: 1.5em 8px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">下面对上面的代码进行详细解释。ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 1.2em;font-weight: bold;display: table;margin: 4em auto 2em;padding-right: 0.2em;padding-left: 0.2em;background: rgb(15, 76, 129);color: rgb(255, 255, 255);">2.1 add_conditional_edgesingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;margin: 1.5em 8px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">首先，我们知道了可以通过add_conditional_edges来对边进行条件添加。这部分代码如下：graph.add_conditional_edges("oracle",router,{
"multiply":"multiply",
"end":END,
})
add_conditional_edges接收三个参数：
如上面的代码，意思就是往 “oracle” node上添加边，这个node有两条边，一条是往“multiply” node上走，一条是往“END”上走。怎么决定往哪个方向去：条件是 router（后面解释），如果 router 返回的是“multiply”，则往“multiply”方向走，如果 router 返回的是 “end”，则走“END”。
来看下这个函数的源码：
defadd_conditional_edges(
self,
start_key:str,
condition:Callable[...,str],
conditional_edge_mapping:Optional[Dict[str,str]]=None,
)->None:
ifself.compiled:
logger.warning(
"Addinganedgetoagraphthathasalreadybeencompiled.Thiswill"
"notbereflectedinthecompiledgraph."
)
ifstart_keynotinself.nodes:
raiseValueError(f"Needtoadd_node`{start_key}`first")
ifiscoroutinefunction(condition):
raiseValueError("Conditioncannotbeacoroutinefunction")
ifconditional_edge_mappingandset(
conditional_edge_mapping.values()
).difference([END]).difference(self.nodes):
raiseValueError(
f"Missingnodeswhichareinconditionaledgemapping.Mapping"
f"containspossibledestinations:"
f"{list(conditional_edge_mapping.values())}.Possiblenodesare"
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”。
defrouter(stateist[BaseMessage]):
tool_calls=state[-1].additional_kwargs.get("tool_calls",[])
iflen(tool_calls):
return"multiply"
else:
return"end"
2.3 各node的定义
（1）起始node：oracle
@tool
defmultiply(first_number:int,second_number:int):
"""Multipliestwonumberstogether."""
returnfirst_number*second_number
model=ChatOpenAI(temperature=0)
model_with_tools=model.bind(tools=[convert_to_openai_tool(multiply)])
graph=MessageGraph()
definvoke_model(stateist[BaseMessage]):
returnmodel_with_tools.invoke(state)
graph.add_node("oracle",invoke_model)
这个node是一个带有Tools 的 ChatOpenAI。在LangChain中使用Tools的详细教程请看这篇文章：【AI大模型应用开发】【LangChain系列】5. LangChain入门：智能体Agents模块的实战详解。简单解释就是：这个node的执行结果，将返回是否应该使用绑定的Tools。
（2）multiply
definvoke_tool(stateist[BaseMessage]):
tool_calls=state[-1].additional_kwargs.get("tool_calls",[])
multiply_call=None
fortool_callintool_calls:
iftool_call.get("function").get("name")=="multiply":
multiply_call=tool_call
ifmultiply_callisNone:
raiseException("Noadderinputfound.")
res=multiply.invoke(
json.loads(multiply_call.get("function").get("arguments"))
)
returnToolMessage(
tool_call_id=multiply_call.get("id"),
content=res
)
graph.add_node("multiply",invoke_tool)
这个node的作用就是执行Tools。
2.4 总体流程
| 欢迎光临 链载Ai (https://www.lianzai.com/) |
Powered by Discuz! X3.5 |