第 9 章 复用与封装：子图

from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END

# ---- 共享的状态 schema：父图与子图都用它 ----
class ScreenState(TypedDict):
    resume_text: str
    candidate: dict
    score: Annotated[int, operator.add]
    breakdown: Annotated[list[str], operator.add]
    report: str                       # 由报告子图填写

# ---- 子图：报告生成（复用单元） ----
def outline(state: ScreenState) -> dict:
    return {"breakdown": ["报告大纲：总分 + 明细 + 建议"]}

def compose(state: ScreenState) -> dict:
    report = (
        f"候选人 {state['candidate']['name']} 总分 {state['score']}\n"
        + "\n".join(state["breakdown"])
    )
    return {"report": report}

report_builder = StateGraph(ScreenState)
report_builder.add_node("outline", outline)
report_builder.add_node("compose", compose)
report_builder.add_edge(START, "outline")
report_builder.add_edge("outline", "compose")
report_builder.add_edge("compose", END)
report_graph = report_builder.compile()      # 子图：编译一次，处处复用

# ---- 父图：主评估流程 ----
def evaluate(state: ScreenState) -> dict:
    c = state["candidate"]
    score = min(100, 20 + len(c["skills"]) * 10 + c["years"] * 3)
    return {"score": score, "breakdown": ["技能匹配良好"]}

builder = StateGraph(ScreenState)
builder.add_node("evaluate", evaluate)
builder.add_node("generate_report", report_graph)   # 直接把子图当节点注册！
builder.add_edge(START, "evaluate")
builder.add_edge("evaluate", "generate_report")
builder.add_edge("generate_report", END)
main_graph = builder.compile()

result = main_graph.invoke({
    "resume_text": "...",
    "candidate": {"name": "李思远", "skills": ["Python", "FastAPI", "PostgreSQL"], "years": 6},
})
print(result["report"])
# 候选人 李思远 总分 68
# 技能匹配良好
# 报告大纲：总分 + 明细 + 建议

from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END

# ---- 子图：完全独立的状态 schema（黑盒） ----
class ReportInput(TypedDict):
    name: str
    score: int
    highlights: list[str]

class ReportState(TypedDict):        # 子图内部状态
    name: str
    score: int
    highlights: list[str]
    report: str

def compose(state: ReportState) -> dict:
    report = f"{state['name']} 总分 {state['score']}：" + "；".join(state["highlights"])
    return {"report": report}

report_builder = StateGraph(ReportState)
report_builder.add_node("compose", compose)
report_builder.add_edge(START, "compose")
report_builder.add_edge("compose", END)
report_graph = report_builder.compile()

# ---- 包装函数：父子状态翻译层 ----
def generate_report(state: ScreenState) -> dict:
    # 父图状态 → 子图输入
    sub_input = {
        "name": state["candidate"]["name"],
        "score": state["score"],
        "highlights": state.get("breakdown", []),
    }
    sub_result = report_graph.invoke(sub_input)     # 子图作为普通 Runnable 调用
    # 子图输出 → 父图补丁
    return {"report": sub_result["report"]}

# ---- 父图：与 9.3 相同的组装方式 ----
builder = StateGraph(ScreenState)
builder.add_node("evaluate", evaluate)              # evaluate 与 9.3 相同
builder.add_node("generate_report", generate_report)  # 注册的是包装函数
builder.add_edge(START, "evaluate")
builder.add_edge("evaluate", "generate_report")
builder.add_edge("generate_report", END)
main_graph = builder.compile()

result = main_graph.invoke({
    "resume_text": "...",
    "candidate": {"name": "李思远", "skills": ["Python", "FastAPI", "PostgreSQL"], "years": 6},
})
print(result["report"])
# 李思远 总分 68：技能匹配良好

from langgraph.types import Command

def subgraph_node(state) -> Command:
    return Command(
        update={"report": "..."},
        goto="next_main_node",              # 目标是父图的节点名
        graph=Command.PARENT,               # 关键：跳转作用于父图层
    )