LangGraph实战：用GPT-4o构建多语言问答系统的完整流程（附代码）

最近在搭建一个面向全球用户的智能客服原型时，我遇到了一个挺有意思的挑战：如何让一个AI系统不仅能理解中文提问，还能自动识别用户语言并用对应语言回答？市面上现成的方案要么太笨重，要么定制化程度不够。折腾了一圈，最后发现用LangGraph来编排整个流程，配合GPT-4o的多语言能力，居然能非常优雅地解决这个问题。今天我就把整个从零搭建的过程，包括踩过的坑和优化技巧，完整地分享出来。

这个方案的核心思路并不复杂，但实现起来有几个关键点需要处理好。首先，系统需要自动检测输入问题的语言；其次，根据检测结果，可能需要调用翻译节点将问题转为系统处理语言（比如英文）；然后，核心的LLM节点生成答案；最后，再根据用户原始语言将答案翻译回去。整个流程涉及多个有状态的节点和条件判断，用传统的线性链（Chain）写起来会非常别扭，而LangGraph的图结构正好能清晰地描述这种工作流。

1. 环境搭建与核心依赖配置

开始写代码之前，得先把环境收拾利索。我习惯用uv来管理Python项目和依赖，它比pip快不少，也能更好地处理版本冲突。如果你还没用过，强烈建议试试。

首先创建一个新项目目录并初始化：

mkdir multilingual-qa-system && cd multilingual-qa-system
uv init

接下来是安装依赖。这里我们主要用到三个库：langgraph、langchain-openai（用于调用GPT-4o），以及langdetect（用于语言检测）。注意版本兼容性，我用的组合是经过验证的。

uv add langgraph==0.2.35 langchain-openai==0.1.0 langdetect==1.0.9 python-dotenv==1.0.0

python-dotenv用来管理环境变量，特别是你的OpenAI API密钥，千万别硬编码在代码里。在项目根目录创建一个.env文件：

OPENAI_API_KEY=你的实际API密钥

然后在代码开头通过dotenv加载它：

import os
from dotenv import load_dotenv

load_dotenv()
assert os.environ.get("OPENAI_API_KEY"), "请确保在 .env 文件中设置了 OPENAI_API_KEY"

提示：如果你在团队中协作，记得把.env文件加入.gitignore，并通过.env.example文件说明需要哪些环境变量。

依赖装好后，我们来规划一下整个图的结构。一个典型的多语言问答流程可以抽象为以下几个节点：

• 语言检测节点：判断用户输入是哪种语言。
• 路由节点：根据检测结果，决定是直接进入问答，还是先走翻译分支。
• 翻译节点（可选）：将非目标语言（如系统处理语言）的问题翻译过去。
• 问答节点：核心的LLM，用系统处理语言生成答案。
• 回译节点（可选）：将答案翻译回用户提问的语言。

在LangGraph里，我们需要先定义一个“状态”（State）来贯穿这些节点，它就像是一个共享的内存，每个节点都能读取和修改其中的部分数据。

2. 定义图状态与构建基础节点

LangGraph的工作流是围绕“状态”流转的。我们需要用一个TypedDict来明确声明这个状态里都有哪些字段，以及它们的类型。这能带来很好的类型提示和错误预防。

假设我们的系统内部处理语言固定为英语，那么状态可能需要包含以下信息：

from typing import TypedDict, Optional, Literal

class GraphState(TypedDict):
    """定义在图节点间传递的状态字典。"""
    # 用户原始输入
    original_question: str
    # 检测到的语言代码，如 'zh-cn', 'en', 'fr'
    detected_language: Optional[str]
    # 经过翻译后，用于系统处理的“标准化”问题（通常是英文）
    processed_question: Optional[str]
    # 系统生成的原始答案（英文）
    raw_answer: Optional[str]
    # 最终返回给用户的答案（用户语言）
    final_answer: Optional[str]
    # 一个标志位，记录当前流程走到了哪一步，方便调试和条件判断
    current_step: Literal["detect", "route", "translate_to_en", "answer", "translate_back", "end"]

有了状态蓝图，就可以开始实现各个节点函数了。每个节点都是一个普通的Python函数，它接收当前的GraphState，执行一些操作，然后返回一个包含更新字段的字典。

第一个节点：语言检测。 这里我用langdetect，它轻量且对短文本效果不错。注意，对于非常短的或混合语言的查询，检测可能不准，所以我在代码里加了个简单回退。

from langdetect import detect, DetectorFactory, LangDetectException

# 确保结果可复现（非必需，但推荐）
DetectorFactory.seed = 0

def detect_language_node(state: GraphState) -> dict:
    """检测用户输入的语言。"""
    question = state["original_question"]
    step = "detect"

    try:
        # langdetect 返回的是双字母代码，如 'zh-cn', 'en', 'ja'
        lang_code = detect(question)
    except LangDetectException:
        # 如果检测失败，默认视为英文
        lang_code = "en"

    # 简单映射，将一些变体统一
    if lang_code.startswith('zh'):
        lang_code = 'zh-cn'
    elif lang_code == 'ko':
        lang_code = 'ko'
    elif lang_code == 'ja':
        lang_code = 'ja'
    else:
        # 其他所有语言暂时按英文处理（或可根据业务扩展）
        lang_code = 'en'

    print(f"[{step}] 检测到语言: {lang_code}, 问题: {question[:50]}...")
    return {"detected_language": lang_code, "current_step": step}

第二个节点：路由决策。 这个节点根据检测到的语言，决定下一步是直接问答，还是需要先翻译。这是一个“条件边”的典型应用场景。

def route_by_language_node(state: GraphState) -> dict:
    """根据检测到的语言，决定下一步走向。"""
    step = "route"
    lang = state["detected_language"]
    question = state["original_question"]

    # 如果检测到的语言已经是英文，或者问题很短（可能检测不准），则直接进入问答节点
    if lang == 'en' or len(question.split()) < 3:
        print(f"[{step}] 语言为英文或问题过短，直接进入问答节点。")
        # 将原始问题直接作为待处理问题
        return {"processed_question": question, "current_step": step, "next_node": "answer_node"}
    else:
        print(f"[{step}] 语言为 {lang}，需要先翻译为英文。")
        # 标记下一步需要进入翻译节点
        return {"current_step": step, "next_node": "translate_to_en_node"}

注意这里我返回了一个next_node字段。这不是GraphState中预定义的，但它可以作为节点内部逻辑的产物，用于指导后续的条件边（Conditional Edge）判断。在LangGraph中，条件边函数正是通过检查状态中的某个值来决定下一个节点的。

3. 实现条件边与LLM集成

LangGraph最强大的特性之一就是能轻松实现分支和循环。我们上面在路由节点里设置了next_node，现在就需要一个条件边函数来读取这个决定。

def decide_next_node(state: GraphState) -> str:
    """条件边函数，决定下一个执行的节点。"""
    # 从状态中读取路由节点设置的目标
    next_node = state.get("next_node")
    if next_node == "answer_node":
        return "answer_node"
    elif next_node == "translate_to_en_node":
        return "translate_to_en_node"
    else:
        # 默认情况下，或者没有明确指示时，结束流程
        return "__end__"

现在来实现两个核心的LLM节点：翻译节点和问答节点。这里我使用langchain-openai来调用GPT-4o，它的多语言理解和生成能力非常出色。为了控制成本和质量，提示词（Prompt）的设计很关键。

首先，初始化LLM客户端：

from langchain_openai import ChatOpenAI

# 使用 GPT-4o，温度设为0保证答案稳定性，可根据场景调整
llm = ChatOpenAI(model="gpt-4o", temperature=0, api_key=os.getenv("OPENAI_API_KEY"))

翻译节点：将用户的问题从任何语言翻译成英文。

from langchain_core.prompts import ChatPromptTemplate

def translate_to_english_node(state: GraphState) -> dict:
    """将非英文问题翻译成英文。"""
    step = "translate_to_en"
    question = state["original_question"]
    src_lang = state["detected_language"]

    # 构建翻译提示词
    translation_prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一位专业的翻译助手。请将用户的问题从{src_lang}准确、流畅地翻译成英文。只输出翻译结果，不要添加任何解释。"),
        ("human", "{question}")
    ])
    # 格式化消息
    messages = translation_prompt.format_messages(src_lang=src_lang, question=question)
    # 调用LLM
    response = llm.invoke(messages)

    translated_text = response.content.strip()
    print(f"[{step}] 翻译完成: 『{question}』 -> 『{translated_text}』")
    return {"processed_question": translated_text, "current_step": step}

问答节点：用英文处理问题，并生成英文答案。这里我设计了一个简单的系统提示，让它扮演一个知识渊博的助手。

def answer_question_node(state: GraphState) -> dict:
    """核心问答节点，用英文处理问题并生成答案。"""
    step = "answer"
    # 使用处理过的问题（可能是翻译后的，也可能是原始的英文问题）
    question_to_answer = state["processed_question"]

    qa_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个乐于助人且准确的AI助手。请用英文回答用户的问题。
        如果问题涉及你不知道或不确定的信息，请诚实说明。
        回答应清晰、有条理，并尽可能提供有帮助的信息。"""),
        ("human", "{question}")
    ])

    messages = qa_prompt.format_messages(question=question_to_answer)
    response = llm.invoke(messages)

    english_answer = response.content.strip()
    print(f"[{step}] 生成英文答案: {english_answer[:100]}...")
    return {"raw_answer": english_answer, "current_step": step}

问答节点结束后，流程还没完。如果用户最初用的是非英文提问，我们还需要把英文答案翻译回用户的语言。这需要另一个条件判断：如果原始语言是英文，则直接结束；否则，进入回译节点。

def should_translate_back(state: GraphState) -> str:
    """判断是否需要将答案翻译回用户语言。"""
    if state["detected_language"] == 'en':
        # 用户用英文问，答案也是英文，直接结束
        return "__end__"
    else:
        # 需要回译
        return "translate_back_node"

回译节点：将英文答案翻译回检测到的用户语言。

def translate_back_node(state: GraphState) -> dict:
    """将英文答案翻译回用户原始语言。"""
    step = "translate_back"
    english_answer = state["raw_answer"]
    target_lang = state["detected_language"]

    back_translation_prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一位专业的翻译助手。请将以下英文文本准确、自然地翻译成{target_lang}。只输出翻译结果，不要添加任何解释。"),
        ("human", "{text}")
    ])

    messages = back_translation_prompt.format_messages(target_lang=target_lang, text=english_answer)
    response = llm.invoke(messages)

    final_text = response.content.strip()
    print(f"[{step}] 回译完成: 『{english_answer[:50]}...』 -> 『{final_text[:50]}...』")
    return {"final_answer": final_text, "current_step": step}

至此，所有功能节点都定义好了。它们各自独立，只通过GraphState交换数据，这种松耦合的设计让后续调试和扩展变得非常方便。

4. 组装图与编译执行

有了节点和边逻辑，现在可以用StateGraph把它们像拼乐高一样组装起来。这一步是LangGraph的核心，它让整个工作流变得可视化且易于管理。

from langgraph.graph import StateGraph, START, END

# 1. 创建图构建器，并指定状态结构
workflow = StateGraph(GraphState)

# 2. 添加所有节点
workflow.add_node("detect_language", detect_language_node)
workflow.add_node("route", route_by_language_node)
workflow.add_node("translate_to_en", translate_to_english_node)
workflow.add_node("answer", answer_question_node)
workflow.add_node("translate_back", translate_back_node)

# 3. 设置起始边
workflow.add_edge(START, "detect_language")

# 4. 添加固定边（无条件流转）
workflow.add_edge("detect_language", "route")
workflow.add_edge("translate_to_en", "answer") # 翻译完后必然去问答
workflow.add_edge("translate_back", END) # 回译完后结束

# 5. 添加条件边（关键！）
# 从 route 节点出来，根据其设置的 next_node 决定去向
workflow.add_conditional_edges(
    "route",
    decide_next_node, # 上面定义的条件判断函数
    {
        "answer_node": "answer", # 如果返回"answer_node"，则跳转到answer节点
        "translate_to_en_node": "translate_to_en", # 如果返回"translate_to_en_node"，则跳转到translate_to_en节点
        "__end__": END # 理论上路由节点不会直接结束，但这里作为兜底
    }
)

# 从 answer 节点出来，判断是否需要回译
workflow.add_conditional_edges(
    "answer",
    should_translate_back, # 上面定义的条件判断函数
    {
        "translate_back_node": "translate_back",
        "__end__": END
    }
)

# 6. 编译图
graph = workflow.compile()

编译成功后，graph对象就是一个可执行的工作流。我们可以用graph.get_graph().draw_mermaid_png()来生成可视化图（如果你安装了pygraphviz），直观地看到整个流程。更简单的方式是直接调用invoke方法运行它。

# 测试中文输入
chinese_result = graph.invoke({"original_question": "量子计算的主要原理是什么？", "current_step": "start"})
print("\n=== 最终答案（中文） ===")
print(chinese_result.get("final_answer", chinese_result.get("raw_answer", "无答案")))

# 测试英文输入
english_result = graph.invoke({"original_question": "Explain the theory of relativity in simple terms.", "current_step": "start"})
print("\n=== 最终答案（英文） ===")
print(english_result.get("final_answer", english_result.get("raw_answer", "无答案")))

运行这段代码，你会在控制台看到每个节点的执行日志，清晰地展示状态是如何流转的。对于中文问题，它会走“检测 -> 路由 -> 翻译 -> 问答 -> 回译”的完整路径；对于英文问题，则会走“检测 -> 路由 -> 问答”的捷径。这种动态路由正是基于图的智能工作流的优势。

5. 高级技巧与生产环境优化

上面的基础版本已经能跑了，但要用于实际项目，还得考虑更多。比如错误处理、流式输出、状态持久化、以及成本控制。

错误处理与重试：网络调用或LLM API可能失败。我们可以用tenacity库为LLM调用添加自动重试。

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def safe_llm_invoke(messages):
    """带重试机制的LLM调用。"""
    return llm.invoke(messages)

# 然后在 translate_to_english_node 等函数中，用 safe_llm_invoke 替换 llm.invoke

流式输出：对于较长的答案，让用户逐步看到结果体验更好。LangGraph支持流式输出每个节点的结果。

# 使用 stream 方法而非 invoke
inputs = {"original_question": "请详细说明机器学习中的过拟合现象。", "current_step": "start"}
for chunk in graph.stream(inputs):
    node_name = list(chunk.keys())[0]
    state_update = chunk[node_name]
    print(f"\n--- 节点 [{node_name}] 更新了状态 ---")
    for key, val in state_update.items():
        if val and key in ['final_answer', 'raw_answer']:
            print(f"{key}: {val}")

状态检查点与持久化：对于长时间运行或多轮对话的Agent，LangGraph内置了检查点（Checkpoint）机制，可以随时保存和恢复图的状态。这对于实现“暂停-继续”或故障恢复非常有用。你需要配置一个持久化后端，比如内存、文件系统或数据库。

from langgraph.checkpoint.memory import MemorySaver

# 创建内存检查点管理器
memory = MemorySaver()
# 在编译图时传入
config = {"configurable": {"thread_id": "user_123"}} # 为每个会话指定唯一thread_id
compiled_graph_with_checkpoints = workflow.compile(checkpointer=memory)

# 第一次调用，状态会被保存
result1 = compiled_graph_with_checkpoints.invoke(
    {"original_question": "第一个问题"},
    config=config
)
# 模拟中途停止...
# 第二次调用，可以从上次的状态继续（假设我们记录了上次结束时的状态ID）
# 实际上，LangGraph会自动管理线程内的状态演进。

成本与延迟优化：每次调用都经过两个LLM节点（问答+可能两次翻译）成本不低。有几个优化思路：

1. 缓存：对常见、重复的问题答案进行缓存。可以用langchain.cache配合SQLiteCache或RedisCache。
2. 短路设计：对于非常简单的、有固定答案的问题（如问候语），可以在路由节点后直接返回预设答案，绕过LLM。
3. 模型选择：不一定所有节点都用GPT-4o。翻译任务用gpt-3.5-turbo可能就足够了，成本更低。可以在不同节点配置不同的LLM模型。

最后，分享一个我实际部署时用的配置表，对比了不同场景下的节点模型选择策略：

节点类型	推荐模型	温度设置	说明
语言检测	langdetect库	-	轻量本地库，零成本，速度快。
翻译节点	gpt-3.5-turbo	0.1	翻译任务相对简单，3.5足够且便宜。追求高质量可用gpt-4o-mini。
核心问答节点	gpt-4o	0~0.3	核心知识生成，需要最强的理解和生成能力。
简单问答/路由	gpt-4o-mini	0	用于判断意图、分类等简单逻辑任务，性价比高。

把这些优化点加上，整个系统就从一个原型进化成了一个健壮、高效、可维护的生产级应用雏形。LangGraph的模块化设计让这种迭代变得非常顺畅，你可以随时替换一个节点，或者增加新的分支流程，而不用担心破坏其他部分。