作者：、、、、南山小雨、、、、
多智能体工作流：串行，并行，一级一个模型负责生成，一个模型负责评估，评估不过继续反馈生成，直到评估通过
多Agent的特点：
多智能体架构：
在新版本的langchain，把langgrach单独拿出了，并赋予了更多的agent功能，langGraph相对于langChain增强了三个部分，其中包括云平台/监看。
节点 边 图 是langGrach中的重要概念，其中边是连接节点的条件：
要点：
1.每个节点的返回return必须是状态。
2.在一开始就规划好全部的节点图，除了conditional，其他的都是确定的，只能执行节点里有的流程。
3.跟langchain不同，大模型不能直接调用工具，只能在AIMessage中包含tools_call，在里面放入想要调用的工具，需要自行用conditional的边来执行。
4.工具只有加入成为一个节点才能最终被执行，多个工具一般为为一个节点。由大语言模型中的tools_call决定调用哪一个。执行到工具的节点，如action（ToolNode），它是根据 last_message.tool_calls 里的 name 来决定执行哪个工具。
注意：
model.bind_tools(tools)工具如果绑定模型，模型就有可能在tools_call中表明调他，但如果它不在节点中，无法被执行。- from langchain_core.messages importAnyMessage
- from typing_extensions importTypedDict
- # 自定义节点间通讯的消息类型
- classState(TypedDict):
- messages: list[AnyMessage]
- extra_field:int
- #定义节点
- from langchain_core.messages importAIMessage
- def node(state: State):
- messages = state["messages"]
- new_message =AIMessage("你好!")return{"messages": messages +[new_message],"extra_field":10}
- from langgraph.graph importStateGraph
- graph_builder =StateGraph(State) #graph_builder就行一张空白的纸，后面的节点都要画在这个纸上 State定义了机器能传递什么数据
- graph_builder.add_node(node) #向图添加节点
- graph_builder.set_entry_point("node") #设置这个图运行的起点
- graph = graph_builder.compile() #编译这个图，让他成为真正可以运行的机器
- #使用：
- # 运行图（最常见方式）
- result = graph.invoke({"messages":[], # 向这个图中添加State类型的数据
- "extra_field":0}) #它就会把这个函数传递给node函数了
- print(result["messages"])
- from typing_extensions importTypedDict
- from IPython.display importImage, display
- from langgraph.graph importSTART, StateGraph
- classState(TypedDict):
- value_1: str
- value_2:int
- def step_1(state: State):return{"value_1":"a"}
- def step_2(state: State):
- current_value_1 = state["value_1"]return{"value_1": f"{current_value_1} b"}
- def step_3(state: State):return{"value_2":10}//一直有一个公共的State值，每个节点都在修改//这里的核心在于，每个节点的返回值都是在修改这个公共的值，下一个节点的输入值，就是这个公共的值
- graph_builder =StateGraph(State)#Add nodes
- graph_builder.add_node(step_1)
- graph_builder.add_node(step_2)
- graph_builder.add_node(step_3)#Add edges
- graph_builder.add_edge(START,"step_1")
- graph_builder.add_edge("step_1","step_2")
- graph_builder.add_edge("step_2","step_3")
- graph = graph_builder.compile()display(Image(graph.get_graph().draw_mermaid_png()))
- graph.invoke({"value_1":"c"}){'value_1':'a b','value_2':10}
- importoperator
- from typing importAnnotated, Any
- from typing_extensions importTypedDict
- from langgraph.graph importStateGraph, START, END
- from IPython.display importImage, display
- #Annotated允许为类型提示添加额外的元数据，而不影响类型检查器对类型本身的理解classState(TypedDict):
- aggregate: Annotated[list,operator.add] #每次都是添加进list
- def a(state: State):print(f'Adding "A" to {state["aggregate"]}')return{"aggregate":["A"]}
- def b(state: State):print(f'Adding "B" to {state["aggregate"]}')return{"aggregate":["B"]}
- def c(state: State):print(f'Adding "C" to {state["aggregate"]}')return{"aggregate":["C"]}
- def d(state: State):print(f'Adding "D" to {state["aggregate"]}')return{"aggregate":["D"]}
- builder =StateGraph(State)
- builder.add_node(a)
- builder.add_node(b)
- builder.add_node(c)
- builder.add_node(d)
- builder.add_edge(START,"a")
- builder.add_edge("a","b")
- builder.add_edge("a","c")
- builder.add_edge("b","d")
- builder.add_edge("c","d")
- builder.add_edge("d", END)
- graph = builder.compile()display(Image(graph.get_graph().draw_mermaid_png()))//运行时配置，{"configurable": {"thread_id": "foo"}} 这个是标记这个graph的，用于Memory（记忆）Checkpointer（断点恢复）。在这里没真正使用它的功能
- graph.invoke({"aggregate":[]},{"configurable":{"thread_id":"foo"}})
- Adding "A" to []
- Adding "B" to ['A']
- Adding "C" to ['A']
- Adding "D" to ['A','B','C']{'aggregate':['A','B','C','D']}
条件分支
- importoperator
- from typing importAnnotated, Literal
- from typing_extensions importTypedDict
- from langgraph.graph importStateGraph, START, END
- classState(TypedDict):
- aggregate: Annotated[list,operator.add]
- def a(state: State):print(f'Node A sees {state["aggregate"]}')return{"aggregate":["A"]}
- def b(state: State):print(f'Node B sees {state["aggregate"]}')return{"aggregate":["B"]}#Define nodes
- builder =StateGraph(State)
- builder.add_node(a)
- builder.add_node(b)
- #这个地方的核心在于return"b"就是直接转到执行b节点
- #Define edges
- def route(state: State)-> Literal["b", END]:iflen(state["aggregate"])<7:return"b"else:return END
- builder.add_edge(START,"a")
- builder.add_conditional_edges("a", route)
- builder.add_edge("b","a")
- graph = builder.compile()
- from IPython.display importImage, display
- display(Image(graph.get_graph().draw_mermaid_png()))
- graph.invoke({"aggregate":[]})
- Node A sees []
- Node B sees ['A']
- Node A sees ['A','B']
- Node B sees ['A','B','A']
- Node A sees ['A','B','A','B']
- Node B sees ['A','B','A','B','A']
- Node A sees ['A','B','A','B','A','B']
- graph.invoke({"aggregate":[]})//为了防止异常情况况下无限循环，设置能在这个结构体上操作的最大次数
- from langgraph.errors importGraphRecursionErrortry:
- graph.invoke({"aggregate":[]},{"recursion_limit":4})
- except GraphRecursionError:print("Recursion Error")
//循环就是添加边的时候，指向值前- importoperator
- from typing importAnnotated, Literal
- from typing_extensions importTypedDict
- from langgraph.graph importStateGraph, START, END
- from IPython.display importImage, display
- classState(TypedDict):
- aggregate: Annotated[list,operator.add]
- def a(state: State):print(f'Node A sees {state["aggregate"]}')return{"aggregate":["A"]}
- def b(state: State):print(f'Node B sees {state["aggregate"]}')return{"aggregate":["B"]}
- def c(state: State):print(f'Node C sees {state["aggregate"]}')return{"aggregate":["C"]}
- def d(state: State):print(f'Node D sees {state["aggregate"]}')return{"aggregate":["D"]}
- # 节点
- builder =StateGraph(State)
- builder.add_node(a)
- builder.add_node(b)
- builder.add_node(c)
- builder.add_node(d)
- # 边
- def route(state: State)-> Literal["b", END]:iflen(state["aggregate"])<7:return"b"else:return END
- builder.add_edge(START,"a")
- builder.add_conditional_edges("a", route)
- builder.add_edge("b","c")
- builder.add_edge("b","d")
- builder.add_edge(["c","d"],"a")
- graph = builder.compile()display(Image(graph.get_graph().draw_mermaid_png()))
- result = graph.invoke({"aggregate":[]})
- Node A sees []
- Node B sees ['A']
- Node C sees ['A','B']
- Node D sees ['A','B']
- Node A sees ['A','B','C','D']
- Node B sees ['A','B','C','D','A']
- Node C sees ['A','B','C','D','A','B']
- Node D sees ['A','B','C','D','A','B']
- Node A sees ['A','B','C','D','A','B','C','D']
- importoperator
- from typing importAnnotated, Sequence
- from typing_extensions importTypedDict
- from langchain_deepseek importChatDeepSeek
- from langchain_openai importChatOpenAIimportos
- from langchain_core.messages importBaseMessage, HumanMessage
- from langchain_core.runnables.config importRunnableConfig
- from langgraph.graph importEND, StateGraph, START
- model =ChatDeepSeek(
- model="Pro/deepseek-ai/DeepSeek-V3",
- temperature=0,
- api_key=os.environ.get("DEEPSEEK_API_KEY"),
- base_url=os.environ.get("DEEPSEEK_API_BASE"),)
- model1 =ChatOpenAI(
- model="gpt-3.5-turbo",
- temperature=0,
- api_key=os.environ.get("OPENAI_API_KEY"),
- base_url=os.environ.get("OPENAI_API_BASE"),)
- # 定义要切换的模型
- models ={"deepseek": model,"openai": model1,}classAgentState(TypedDict):
- messages: Annotated[Sequence[BaseMessage],operator.add]
- def _call_model(state: AgentState, config: RunnableConfig):
- # 使用LCEL的配置
- model_name = config["configurable"].get("model","deepseek")
- model = models[model_name]
- response = model.invoke(state["messages"])return{"messages":[response]}#Define anew graph
- builder =StateGraph(AgentState)
- builder.add_node("model", _call_model)
- builder.add_edge(START,"model")
- builder.add_edge("model", END)
- graph = builder.compile()//没有增加运行时配置的情况下，它会默认调用deepseek
- graph.invoke({"messages":[HumanMessage(content="hi 你是谁？")]})//图的运行时配置，就是多个行参而已
- config ={"configurable":{"model":"openai"}}
- graph.invoke({"messages":[HumanMessage(content="hi 你是谁？")]}, config=config)
[code]from langchain_deepseek importChatDeepSeekimportos
from langgraph.graph importStateGraph, MessagesState, START
model =ChatDeepSeek(
model="Pro/deepseek-ai/DeepSeek-V3",
temperature=0,
api_key=os.environ.get("DEEPSEEK_API_KEY"),
base_url=os.environ.get("DEEPSEEK_API_BASE"),)
def call_model(state: MessagesState):
response = model.invoke(state["messages"])return{"messages": response}
builder =StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START,"call_model")
from langgraph.checkpoint.memory importMemorySaver
# 使用 MemorySaver 保存中间状态 这个是内存，且聊天记录是没有向量化，直接存入
memory =MemorySaver()
#加上memory就能保持记忆了，激活持久化层
graph = builder.compile(checkpointer=memory)
config ={"configurable":{"thread_id":"1"}}
input_message ={"role":"user","content":"hi! 我是tomie"}#stream_mode="values"方式就是每次经过一个节点就返回一次MessagesState数据for chunk in graph.stream({"messages":[input_message]}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
input_message ={"role":"user","content":"我叫什么名字?"}for chunk in graph.stream({"messages":[input_message]}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()================================ Human Message =================================
我叫什么名字?================================== Ai Message ==================================
哈哈，你刚刚说过啦！你叫 **Tomie**～ |
广告