LangGraph子图可控性

示例代码

注：将ppt后缀名改为zip后解压

python版本：3.12.4

​

子图可控性

在复杂的多代理协作场景中，LangGraph 支持子图之间的状态共享。主图可以通过状态对象与子图交互，确保各子图能够访问和更新全局状态。这种机制使得多个智能体可以协同工作，同时保持状态的一致性。

对于更复杂的系统，子图是一个有用的设计原则。子图允许您在图的不同部分创建和管理不同的状态。

%pip install -U langgraph

示例：我有一个接受日志并执行两个独立子任务的系统。首先，它将对它们进行总结。其次，它将总结日志中捕获的任何故障模式。我想在两个不同的子图中执行这两个操作。

最重要的是要认识到图之间的信息传递。入口图 是父级，两个子图中的每一个都被定义为 入口图 中的节点。两个子图都继承了父级 入口图 的状态；我可以在每个子图中访问 docs，只需在子图状态中指定它（参见图）。每个子图都可以有自己的私有状态。任何我想传播回父级 入口图 的值（用于最终报告）只需在我的 入口图 状态中定义即可（例如，summary report 和 failure report）。

#示例：sub_graph.py
from typing import List, TypedDict, Optional
from langgraph.graph import StateGraph, START, END

# 定义日志的结构
class Logs(TypedDict):
    id: str  # 日志的唯一标识符
    question: str  # 问题文本
    docs: Optional[List]  # 可选的相关文档列表
    answer: str  # 回答文本
    grade: Optional[int]  # 可选的评分
    grader: Optional[str]  # 可选的评分者
    feedback: Optional[str]  # 可选的反馈信息

# 定义故障分析状态的结构
class FailureAnalysisState(TypedDict):
    docs: List[Logs]  # 日志列表
    failures: List[Logs]  # 失败的日志列表
    fa_summary: str  # 故障分析总结

# 获取失败的日志
def get_failures(state):
    docs = state["docs"]  # 从状态中获取日志
    failures = [doc for doc in docs if "grade" in doc]  # 筛选出包含评分的日志
    return {"failures": failures}  # 返回包含失败日志的字典

# 生成故障分析总结
def generate_summary(state):
    failures = state["failures"]  # 从状态中获取失败的日志
    # 添加函数：fa_summary = summarize(failures)
    fa_summary = "Poor quality retrieval of Chroma documentation."  # 固定的总结内容
    return {"fa_summary": fa_summary}  # 返回包含总结的字典

# 创建故障分析的状态图
fa_builder = StateGraph(FailureAnalysisState)
fa_builder.add_node("get_failures", get_failures)  # 添加节点：获取失败的日志
fa_builder.add_node("generate_summary", generate_summary)  # 添加节点：生成总结
fa_builder.add_edge(START, "get_failures")  # 添加边：从开始到获取失败的日志
fa_builder.add_edge("get_failures", "generate_summary")  # 添加边：从获取失败的日志到生成总结
fa_builder.add_edge("generate_summary", END)  # 添加边：从生成总结到结束

# 定义问题总结状态的结构
class QuestionSummarizationState(TypedDict):
    docs: List[Logs]  # 日志列表
    qs_summary: str  # 问题总结
    report: str  # 报告

# 生成问题总结
def generate_summary(state):
    docs = state["docs"]  # 从状态中获取日志
    # 添加函数：summary = summarize(docs)
    summary = "Questions focused on usage of ChatOllama and Chroma vector store."  # 固定的总结内容
    return {"qs_summary": summary}  # 返回包含总结的字典