AI创想

标题: LangGraph [打印本页]
作者: 创想小编    时间: 3 小时前
标题: LangGraph
作者：、、、、南山小雨、、、、
多智能体工作流：串行，并行，一级一个模型负责生成，一个模型负责评估，评估不过继续反馈生成，直到评估通过

(, 下载次数: 0)

多Agent的特点：

(, 下载次数: 0)


多智能体架构：
(, 下载次数: 0)

在新版本的langchain，把langgrach单独拿出了，并赋予了更多的agent功能，langGraph相对于langChain增强了三个部分，其中包括云平台/监看。
(, 下载次数: 0)

节点 边 图 是langGrach中的重要概念，其中边是连接节点的条件：
(, 下载次数: 0)

要点：
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中表明调他，但如果它不在节点中，无法被执行。
  1. from langchain_core.messages importAnyMessage
  2. from typing_extensions importTypedDict
  4. # 自定义节点间通讯的消息类型
  5. classState(TypedDict):
  6.     messages: list[AnyMessage]
  7.     extra_field:int
  10. #定义节点
  11. from langchain_core.messages importAIMessage
  12. def node(state: State):
  13.     messages = state["messages"]
  14.     new_message =AIMessage("你好!")return{"messages": messages +[new_message],"extra_field":10}
  19. from langgraph.graph importStateGraph
  20. graph_builder =StateGraph(State) #graph_builder就行一张空白的纸，后面的节点都要画在这个纸上 State定义了机器能传递什么数据
  21. graph_builder.add_node(node) #向图添加节点
  22. graph_builder.set_entry_point("node") #设置这个图运行的起点
  23. graph = graph_builder.compile() #编译这个图，让他成为真正可以运行的机器
  25. #使用：
  26. # 运行图（最常见方式）
  27. result = graph.invoke({"messages":[],           # 向这个图中添加State类型的数据
  28.     "extra_field":0}) #它就会把这个函数传递给node函数了
  30. print(result["messages"])
复制代码
串行控制
  1. from typing_extensions importTypedDict
  2. from IPython.display importImage, display
  3. from langgraph.graph importSTART, StateGraph
  5. classState(TypedDict):
  6.     value_1: str
  7.     value_2:int
  9. def step_1(state: State):return{"value_1":"a"}
  11. def step_2(state: State):
  12.     current_value_1 = state["value_1"]return{"value_1": f"{current_value_1} b"}
  14. def step_3(state: State):return{"value_2":10}//一直有一个公共的State值，每个节点都在修改//这里的核心在于，每个节点的返回值都是在修改这个公共的值，下一个节点的输入值，就是这个公共的值
  15. graph_builder =StateGraph(State)#Add nodes
  16. graph_builder.add_node(step_1)
  17. graph_builder.add_node(step_2)
  18. graph_builder.add_node(step_3)#Add edges
  19. graph_builder.add_edge(START,"step_1")
  20. graph_builder.add_edge("step_1","step_2")
  21. graph_builder.add_edge("step_2","step_3")
  23. graph = graph_builder.compile()display(Image(graph.get_graph().draw_mermaid_png()))
复制代码
(, 下载次数: 0)

  1. graph.invoke({"value_1":"c"}){'value_1':'a b','value_2':10}
复制代码
分支结构
  1. importoperator
  2. from typing importAnnotated, Any
  4. from typing_extensions importTypedDict
  6. from langgraph.graph importStateGraph, START, END
  7. from IPython.display importImage, display
  9. #Annotated允许为类型提示添加额外的元数据，而不影响类型检查器对类型本身的理解classState(TypedDict):
  10.     aggregate: Annotated[list,operator.add] #每次都是添加进list
  13. def a(state: State):print(f'Adding "A" to {state["aggregate"]}')return{"aggregate":["A"]}
  16. def b(state: State):print(f'Adding "B" to {state["aggregate"]}')return{"aggregate":["B"]}
  19. def c(state: State):print(f'Adding "C" to {state["aggregate"]}')return{"aggregate":["C"]}
  22. def d(state: State):print(f'Adding "D" to {state["aggregate"]}')return{"aggregate":["D"]}
  25. builder =StateGraph(State)
  26. builder.add_node(a)
  27. builder.add_node(b)
  28. builder.add_node(c)
  29. builder.add_node(d)
  30. builder.add_edge(START,"a")
  31. builder.add_edge("a","b")
  32. builder.add_edge("a","c")
  33. builder.add_edge("b","d")
  34. builder.add_edge("c","d")
  35. builder.add_edge("d", END)
  36. graph = builder.compile()display(Image(graph.get_graph().draw_mermaid_png()))//运行时配置，{"configurable": {"thread_id": "foo"}} 这个是标记这个graph的，用于Memory（记忆）Checkpointer（断点恢复）。在这里没真正使用它的功能
  37. graph.invoke({"aggregate":[]},{"configurable":{"thread_id":"foo"}})
  38. Adding "A" to []
  39. Adding "B" to ['A']
  40. Adding "C" to ['A']
  41. Adding "D" to ['A','B','C']{'aggregate':['A','B','C','D']}
复制代码
(, 下载次数: 0)


条件分支
  1. importoperator
  2. from typing importAnnotated, Literal
  4. from typing_extensions importTypedDict
  6. from langgraph.graph importStateGraph, START, END
  9. classState(TypedDict):
  10.     aggregate: Annotated[list,operator.add]
  13. def a(state: State):print(f'Node A sees {state["aggregate"]}')return{"aggregate":["A"]}
  16. def b(state: State):print(f'Node B sees {state["aggregate"]}')return{"aggregate":["B"]}#Define nodes
  17. builder =StateGraph(State)
  18. builder.add_node(a)
  19. builder.add_node(b)
  21. #这个地方的核心在于return"b"就是直接转到执行b节点
  22. #Define edges
  23. def route(state: State)-> Literal["b", END]:iflen(state["aggregate"])<7:return"b"else:return END
  26. builder.add_edge(START,"a")
  27. builder.add_conditional_edges("a", route)
  28. builder.add_edge("b","a")
  29. graph = builder.compile()
复制代码
//打印
  1. from IPython.display importImage, display
  3. display(Image(graph.get_graph().draw_mermaid_png()))
复制代码
(, 下载次数: 0)

  1. graph.invoke({"aggregate":[]})
  3. Node A sees []
  4. Node B sees ['A']
  5. Node A sees ['A','B']
  6. Node B sees ['A','B','A']
  7. Node A sees ['A','B','A','B']
  8. Node B sees ['A','B','A','B','A']
  9. Node A sees ['A','B','A','B','A','B']
  10. graph.invoke({"aggregate":[]})//为了防止异常情况况下无限循环，设置能在这个结构体上操作的最大次数
  11. from langgraph.errors importGraphRecursionErrortry:
  12.     graph.invoke({"aggregate":[]},{"recursion_limit":4})
  13. except GraphRecursionError:print("Recursion Error")
复制代码
循环

//循环就是添加边的时候，指向值前
  1. importoperator
  2. from typing importAnnotated, Literal
  4. from typing_extensions importTypedDict
  6. from langgraph.graph importStateGraph, START, END
  7. from IPython.display importImage, display
  9. classState(TypedDict):
  10.     aggregate: Annotated[list,operator.add]
  13. def a(state: State):print(f'Node A sees {state["aggregate"]}')return{"aggregate":["A"]}
  16. def b(state: State):print(f'Node B sees {state["aggregate"]}')return{"aggregate":["B"]}
  19. def c(state: State):print(f'Node C sees {state["aggregate"]}')return{"aggregate":["C"]}
  22. def d(state: State):print(f'Node D sees {state["aggregate"]}')return{"aggregate":["D"]}
  25. # 节点
  26. builder =StateGraph(State)
  27. builder.add_node(a)
  28. builder.add_node(b)
  29. builder.add_node(c)
  30. builder.add_node(d)
  33. # 边
  34. def route(state: State)-> Literal["b", END]:iflen(state["aggregate"])<7:return"b"else:return END
  37. builder.add_edge(START,"a")
  38. builder.add_conditional_edges("a", route)
  39. builder.add_edge("b","c")
  40. builder.add_edge("b","d")
  41. builder.add_edge(["c","d"],"a")
  42. graph = builder.compile()display(Image(graph.get_graph().draw_mermaid_png()))
复制代码
(, 下载次数: 0)

  1. result = graph.invoke({"aggregate":[]})
  2. Node A sees []
  3. Node B sees ['A']
  4. Node C sees ['A','B']
  5. Node D sees ['A','B']
  6. Node A sees ['A','B','C','D']
  7. Node B sees ['A','B','C','D','A']
  8. Node C sees ['A','B','C','D','A','B']
  9. Node D sees ['A','B','C','D','A','B']
  10. Node A sees ['A','B','C','D','A','B','C','D']
复制代码
图的运行时配置
  1. importoperator
  2. from typing importAnnotated, Sequence
  3. from typing_extensions importTypedDict
  5. from langchain_deepseek importChatDeepSeek
  6. from langchain_openai importChatOpenAIimportos
  7. from langchain_core.messages importBaseMessage, HumanMessage
  8. from langchain_core.runnables.config importRunnableConfig
  9. from langgraph.graph importEND, StateGraph, START
  11. model =ChatDeepSeek(
  12.     model="Pro/deepseek-ai/DeepSeek-V3",
  13.     temperature=0,
  14.     api_key=os.environ.get("DEEPSEEK_API_KEY"),
  15.     base_url=os.environ.get("DEEPSEEK_API_BASE"),)
  17. model1 =ChatOpenAI(
  18.     model="gpt-3.5-turbo",
  19.     temperature=0,
  20.     api_key=os.environ.get("OPENAI_API_KEY"),
  21.     base_url=os.environ.get("OPENAI_API_BASE"),)
  23. # 定义要切换的模型
  24. models ={"deepseek": model,"openai": model1,}classAgentState(TypedDict):
  25.     messages: Annotated[Sequence[BaseMessage],operator.add]
  28. def _call_model(state: AgentState, config: RunnableConfig):
  29.     # 使用LCEL的配置
  30.     model_name = config["configurable"].get("model","deepseek")
  31.     model = models[model_name]
  32.     response = model.invoke(state["messages"])return{"messages":[response]}#Define anew graph
  33. builder =StateGraph(AgentState)
  34. builder.add_node("model", _call_model)
  35. builder.add_edge(START,"model")
  36. builder.add_edge("model", END)
  38. graph = builder.compile()//没有增加运行时配置的情况下，它会默认调用deepseek
  39. graph.invoke({"messages":[HumanMessage(content="hi 你是谁？")]})//图的运行时配置，就是多个行参而已
  40. config ={"configurable":{"model":"openai"}}
  41. 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**～



欢迎光临 AI创想 (https://llms-ai.com/) Powered by Discuz! X3.4