15、16节的核心知识点是 LangGraph的子图（Subgraph）设计与实现。这是一种将复杂系统拆分为独立、可复用的“子系统”，并通过标准接口进行组合和沟通的强大思想。

为了方便你理解，我把这两节课的核心概念整理在下面的表格里了：

💻 动手实战：构建一个新闻处理工作流

这个案例会模拟一个内容处理应用：父图接收一篇新闻初稿，然后并行调用两个子图——一个负责将英文翻译成中文，另一个负责将中文文本进行情感打分。

1. 环境准备与模型配置

你需要先准备好模型。这个案例支持使用阿里云百炼的Qwen模型或模拟模型进行测试，代码中已经包含了详细的配置说明。

# ---------- 1. 环境准备与模型配置 ----------
import os
from dotenv import load_dotenv
from typing import TypedDict, List, Dict, Any
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

load_dotenv()

def setup_llm():
    """配置Qwen模型或使用模拟模型"""
    # 方式一：使用阿里云百炼Qwen（推荐）
    try:
        llm = ChatOpenAI(
            model="qwen-plus",
            temperature=0.3,
            api_key=os.getenv("DASHSCOPE_API_KEY"),
            base_url=os.getenv("DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1"),
            model_kwargs={"extra_body": {"enable_thinking": False}}
        )
        # 测试连接
        llm.invoke("测试")
        print("✅ 已连接Qwen模型服务\n")
        return llm
    except Exception as e:
        print(f"⚠️ Qwen模型连接失败: {e}")
        print("将使用模拟模型进行演示...\n")
        # 方式二：模拟模型（无需API）
        class MockLLM:
            def invoke(self, prompt):
                content = prompt if isinstance(prompt, str) else prompt[-1].content
                if "translate" in content.lower():
                    return type('', (), {'content': f'[翻译结果] {content.split("text:")[-1].strip()} (中文版)'})()
                elif "sentiment" in content.lower():
                    return type('', (), {'content': 'positive' if "good" in content else 'neutral'})()
                return type('', (), {'content': '处理完成'})()
        return MockLLM()

llm = setup_llm()

2. 定义父图状态

# ---------- 2. 定义父图状态 ----------
class ParentState(TypedDict):
    original_article: str      # 原始英文文章
    translated_article: str     # 中文翻译结果
    sentiment_scores: List[Dict[str, str]]  # 情感分析结果
    current_step: str           # 当前处理步骤
    final_output: str           # 最终输出

3. 创建第一个子图：翻译模块

# ---------- 3. 创建子图 A：翻译模块 ----------
class TranslationSubgraphState(TypedDict):
    input_text: str        # 输入：待翻译文本
    output_text: str       # 输出：翻译结果

def translate_node(state: TranslationSubgraphState) -> TranslationSubgraphState:
    """翻译节点：将英文翻译成中文"""
    text = state["input_text"]
    # 构建翻译提示词
    prompt = f"请将以下英文翻译成中文，只返回翻译结果，不要有任何解释。\n\nEnglish: {text}\n\nChinese:"
    
    # 调用模型
    response = llm.invoke(prompt)
    translation = response.content if hasattr(response, 'content') else str(response)
    
    print(f"    📝 [子图-翻译] 翻译完成: {translation[:50]}...")
    return {"output_text": translation}

# 构建翻译子图
translation_builder = StateGraph(TranslationSubgraphState)
translation_builder.add_node("translate", translate_node)
translation_builder.set_entry_point("translate")
translation_builder.add_edge("translate", END)
translation_subgraph = translation_builder.compile()

# ⚠️ 重要：子图包装函数，转换输入/输出格式
def call_translation_subgraph(state: ParentState) -> ParentState:
    """包装函数：从父图提取输入，调用子图，返回结果更新父图"""
    print("  🔄 [父图] 调用翻译子图...")
    # 输入转换：父图状态 -> 子图状态
    subgraph_input = {"input_text": state["original_article"]}
    # 调用子图
    subgraph_output = translation_subgraph.invoke(subgraph_input)
    # 输出转换：子图状态 -> 父图状态
    return {"translated_article": subgraph_output["output_text"]}

4. 创建第二个子图：情感分析模块

# ---------- 4. 创建子图 B：情感分析模块 ----------
class SentimentSubgraphState(TypedDict):
    text: str              # 输入：待分析文本
    sentiment: str         # 输出：情感标签
    confidence: float      # 输出：置信度

def sentiment_node(state: SentimentSubgraphState) -> SentimentSubgraphState:
    """情感分析节点：分析文本情感倾向"""
    text = state["text"]
    
    # 模拟根据关键词判断情感
    positive_words = ["好", "棒", "喜欢", "优秀", "晴天"]
    negative_words = ["差", "坏", "讨厌", "糟糕", "阴天"]
    
    sentiment = "neutral"
    confidence = 0.7
    
    for word in positive_words:
        if word in text:
            sentiment = "positive"
            confidence = 0.85
            break
    for word in negative_words:
        if word in text:
            sentiment = "negative"
            confidence = 0.80
            break
    
    print(f"    📊 [子图-情感分析] 文本: {text[:30]}... -> 情感: {sentiment} (置信度: {confidence})")
    return {"sentiment": sentiment, "confidence": confidence}

# 构建情感分析子图
sentiment_builder = StateGraph(SentimentSubgraphState)
sentiment_builder.add_node("analyze", sentiment_node)
sentiment_builder.set_entry_point("analyze")
sentiment_builder.add_edge("analyze", END)
sentiment_subgraph = sentiment_builder.compile()

# 子图包装函数
def call_sentiment_subgraph(state: ParentState) -> ParentState:
    """包装函数：调用情感分析子图"""
    print("  🔄 [父图] 调用情感分析子图...")
    subgraph_input = {"text": state["translated_article"]}
    subgraph_output = sentiment_subgraph.invoke(subgraph_input)
    
    # 更新情感分析结果列表
    sentiment_result = {
        "text": state["translated_article"][:50] + "...",
        "sentiment": subgraph_output["sentiment"],
        "confidence": subgraph_output["confidence"]
    }
    
    current_scores = state.get("sentiment_scores", [])
    current_scores.append(sentiment_result)
    
    return {"sentiment_scores": current_scores}

5. 创建父图并集成子图

# ---------- 5. 创建父图并集成子图 ----------
def prepare_input_node(state: ParentState) -> ParentState:
    """准备输入节点"""
    print("📥 [父图] 接收原始文章")
    return {"current_step": "preparing"}

def aggregate_results_node(state: ParentState) -> ParentState:
    """汇总结果节点：整合两个子图的结果"""
    print("\n📊 [父图] 汇总处理结果...")
    output = f"""=== 新闻处理报告 ===

【原始英文原文】
{state["original_article"]}

【中文译文】
{state["translated_article"]}

【情感分析结果】
"""
    for score in state.get("sentiment_scores", []):
        output += f"\n  文本片段: {score['text']}\n  情感: {score['sentiment']} (置信度: {score['confidence']})\n"
    
    print(f"✅ [父图] 处理完成!")
    return {"final_output": output, "current_step": "completed"}

# 构建父图
parent_builder = StateGraph(ParentState)

# 添加节点
parent_builder.add_node("prepare", prepare_input_node)
parent_builder.add_node("translate_article", call_translation_subgraph)      # 集成子图A
parent_builder.add_node("analyze_sentiment", call_sentiment_subgraph)       # 集成子图B
parent_builder.add_node("aggregate", aggregate_results_node)

# 定义流程图
parent_builder.set_entry_point("prepare")
parent_builder.add_edge("prepare", "translate_article")
parent_builder.add_edge("translate_article", "analyze_sentiment")
parent_builder.add_edge("analyze_sentiment", "aggregate")
parent_builder.add_edge("aggregate", END)

# 编译父图
parent_graph = parent_builder.compile()

6. 运行案例并执行测试

# ---------- 6. 运行测试 ----------
def visualize_graph():
    """尝试打印图结构"""
    try:
        print("\n📊 图结构预览:")
        print(parent_graph.get_graph().draw_ascii())
    except Exception as e:
        print(f"无法打印ASCII图: {e}")

if __name__ == "__main__":
    print("🚀 启动LangGraph子图案例演示\n")
    print("="*60)
    
    # 测试新闻内容
    test_article = """Sunny weather expected in Beijing today. 
The temperature will be around 25°C, perfect for outdoor activities. 
The air quality is good, making it a great day to enjoy the city parks."""
    
    print(f"📰 测试内容: {test_article}\n")
    
    # 初始状态
    initial_state = {
        "original_article": test_article,
        "translated_article": "",
        "sentiment_scores": [],
        "current_step": "",
        "final_output": ""
    }
    
    # 执行父图（自动调用子图）
    print("⚙️ 开始执行工作流...\n")
    final_state = parent_graph.invoke(initial_state)
    
    print("\n" + "="*60)
    print("📋 最终结果:\n")
    print(final_state["final_output"])
    
    # 可视化图结构
    # visualize_graph()

📝 核心收获与面试准备

实践收获

通过这个案例，你应该亲自实践了：

• 如何定义两个独立子图，分别负责翻译和情感分析。
• 如何通过包装函数实现输入/输出转换，在保持内部逻辑独立的同时完成数据传递。
• 如何在父图中并行调用两个子图，并最终将结果汇总。

# ============================================================
# LangGraph 子图（Subgraph）案例 - 完整代码
# 功能：父图调用翻译子图 + 情感分析子图，处理新闻文章
# ============================================================

import os
from typing import TypedDict, List, Dict, Any
from dotenv import load_dotenv
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

# 加载 .env 环境变量（如果有）
load_dotenv()

# ================== 1. 模型配置（支持真实Qwen或模拟） ==================
def setup_llm():
    """配置阿里云百炼Qwen模型，若失败则使用模拟模型"""
    try:
        llm = ChatOpenAI(
            model="qwen-plus",
            temperature=0.3,
            api_key=os.getenv("DASHSCOPE_API_KEY"),
            base_url=os.getenv("DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1"),
            model_kwargs={"extra_body": {"enable_thinking": False}}
        )
        # 测试连接
        llm.invoke("测试")
        print("✅ 已连接Qwen模型服务\n")
        return llm
    except Exception as e:
        print(f"⚠️ Qwen模型连接失败: {e}")
        print("将使用模拟模型进行演示...\n")
        # 模拟模型（无需API）
        class MockLLM:
            def invoke(self, prompt):
                content = prompt if isinstance(prompt, str) else prompt[-1].content
                if "translate" in content.lower():
                    # 模拟翻译
                    original = content.split("English:")[-1].split("Chinese:")[0].strip() if "English:" in content else content
                    return type('', (), {'content': f'[模拟翻译] {original} (中文版)'})()
                elif "sentiment" in content.lower():
                    return type('', (), {'content': 'positive' if "good" in content else 'neutral'})()
                return type('', (), {'content': '处理完成'})()
        return MockLLM()

llm = setup_llm()

# ================== 2. 定义父图状态 ==================
class ParentState(TypedDict):
    original_article: str          # 原始英文文章
    translated_article: str        # 中文翻译结果
    sentiment_scores: List[Dict[str, str]]  # 情感分析结果列表
    current_step: str              # 当前处理步骤
    final_output: str              # 最终输出报告

# ================== 3. 子图 A：翻译模块 ==================
class TranslationSubgraphState(TypedDict):
    input_text: str
    output_text: str

def translate_node(state: TranslationSubgraphState) -> TranslationSubgraphState:
    """翻译节点：调用LLM将英文翻译成中文"""
    text = state["input_text"]
    prompt = f"请将以下英文翻译成中文，只返回翻译结果，不要有任何解释。\n\nEnglish: {text}\n\nChinese:"
    response = llm.invoke(prompt)
    translation = response.content if hasattr(response, 'content') else str(response)
    print(f"    📝 [子图-翻译] 翻译完成: {translation[:50]}...")
    return {"output_text": translation}

# 构建翻译子图
translation_builder = StateGraph(TranslationSubgraphState)
translation_builder.add_node("translate", translate_node)
translation_builder.set_entry_point("translate")
translation_builder.add_edge("translate", END)
translation_subgraph = translation_builder.compile()

def call_translation_subgraph(state: ParentState) -> ParentState:
    """包装函数：从父图提取输入，调用翻译子图，返回结果更新父图"""
    print("  🔄 [父图] 调用翻译子图...")
    subgraph_input = {"input_text": state["original_article"]}
    subgraph_output = translation_subgraph.invoke(subgraph_input)
    return {"translated_article": subgraph_output["output_text"]}

# ================== 4. 子图 B：情感分析模块 ==================
class SentimentSubgraphState(TypedDict):
    text: str
    sentiment: str
    confidence: float

def sentiment_node(state: SentimentSubgraphState) -> SentimentSubgraphState:
    """情感分析节点：基于关键词判断情感（模拟）"""
    text = state["text"]
    positive_words = ["好", "棒", "喜欢", "优秀", "晴天", "good", "great", "perfect"]
    negative_words = ["差", "坏", "讨厌", "糟糕", "阴天", "bad", "terrible"]
    
    sentiment = "neutral"
    confidence = 0.7
    text_lower = text.lower()
    for word in positive_words:
        if word in text_lower:
            sentiment = "positive"
            confidence = 0.85
            break
    for word in negative_words:
        if word in text_lower:
            sentiment = "negative"
            confidence = 0.80
            break
    
    print(f"    📊 [子图-情感分析] 文本: {text[:30]}... -> 情感: {sentiment} (置信度: {confidence})")
    return {"sentiment": sentiment, "confidence": confidence}

# 构建情感分析子图
sentiment_builder = StateGraph(SentimentSubgraphState)
sentiment_builder.add_node("analyze", sentiment_node)
sentiment_builder.set_entry_point("analyze")
sentiment_builder.add_edge("analyze", END)
sentiment_subgraph = sentiment_builder.compile()

def call_sentiment_subgraph(state: ParentState) -> ParentState:
    """包装函数：调用情感分析子图"""
    print("  🔄 [父图] 调用情感分析子图...")
    subgraph_input = {"text": state["translated_article"]}
    subgraph_output = sentiment_subgraph.invoke(subgraph_input)
    
    sentiment_result = {
        "text": state["translated_article"][:50] + "...",
        "sentiment": subgraph_output["sentiment"],
        "confidence": subgraph_output["confidence"]
    }
    current_scores = state.get("sentiment_scores", [])
    current_scores.append(sentiment_result)
    return {"sentiment_scores": current_scores}

# ================== 5. 父图节点定义 ==================
def prepare_input_node(state: ParentState) -> ParentState:
    """准备输入节点"""
    print("📥 [父图] 接收原始文章")
    return {"current_step": "preparing"}

def aggregate_results_node(state: ParentState) -> ParentState:
    """汇总结果节点：整合翻译和情感分析结果"""
    print("\n📊 [父图] 汇总处理结果...")
    output = f"""=== 新闻处理报告 ===

【原始英文原文】
{state["original_article"]}

【中文译文】
{state["translated_article"]}

【情感分析结果】
"""
    for score in state.get("sentiment_scores", []):
        output += f"\n  文本片段: {score['text']}\n  情感: {score['sentiment']} (置信度: {score['confidence']})\n"
    
    print("✅ [父图] 处理完成!")
    return {"final_output": output, "current_step": "completed"}

# ================== 6. 构建父图并集成子图 ==================
parent_builder = StateGraph(ParentState)

# 添加节点
parent_builder.add_node("prepare", prepare_input_node)
parent_builder.add_node("translate_article", call_translation_subgraph)   # 子图A
parent_builder.add_node("analyze_sentiment", call_sentiment_subgraph)    # 子图B
parent_builder.add_node("aggregate", aggregate_results_node)

# 定义流程边
parent_builder.set_entry_point("prepare")
parent_builder.add_edge("prepare", "translate_article")
parent_builder.add_edge("translate_article", "analyze_sentiment")
parent_builder.add_edge("analyze_sentiment", "aggregate")
parent_builder.add_edge("aggregate", END)

# 编译父图
parent_graph = parent_builder.compile()

# ================== 7. 运行测试 ==================
if __name__ == "__main__":
    print("🚀 启动 LangGraph 子图案例演示\n")
    print("=" * 60)

    # 测试新闻文章（英文）
    test_article = """Sunny weather expected in Beijing today. 
The temperature will be around 25°C, perfect for outdoor activities. 
The air quality is good, making it a great day to enjoy the city parks."""

    print(f"📰 测试内容:\n{test_article}\n")

    initial_state = {
        "original_article": test_article,
        "translated_article": "",
        "sentiment_scores": [],
        "current_step": "",
        "final_output": ""
    }

    print("⚙️ 开始执行工作流...\n")
    final_state = parent_graph.invoke(initial_state)

    print("\n" + "=" * 60)
    print("📋 最终结果:\n")
    print(final_state["final_output"])

面试常见问题