【001】LangChain + 通义千问大模型实现多语言互译-可运行(完整代码在结尾）

稀稀落落：：

342人浏览 · 2026-01-14 10:04:41

稀稀落落：： · 2026-01-14 10:04:41 发布

零基础吃透这份 LangChain+通义千问翻译代码（全程大白话+图解+实操）

✅ 先搞懂【整体功能】

这份代码的核心作用：封装了一个「专业翻译助手」LLM应用，支持2种调用方式

同步翻译：传入「待翻译文本+目标语言」，一次性返回完整翻译结果（精准、无废话）
流式翻译：传入「待翻译文本+目标语言」，一个字/一个词慢慢返回翻译结果（和ChatGPT打字输出的效果一模一样）
翻译规则：严格只翻译、不添加任何额外解释（比如不会出现“这句话的意思是：xxx”），精准度拉满。

✅ 第一步：环境准备 & 前置知识（必须先做，否则运行报错）

1. 安装依赖包

运行代码前，必须在终端执行下面的命令安装需要的库，缺一不可：

pip install langchain-core dashscope python-dotenv

langchain-core：LangChain的核心库，提供所有基础组件（今天讲的RunnableLambda、StrOutputParser都在这里）
dashscope：阿里云通义千问的官方Python SDK，专门用来调用通义千问的API
python-dotenv：可选，用来管理配置文件的密钥

2. 通义千问API-KEY准备

代码里用到了 DASHSCOPE_API_KEY，这个是调用通义千问的「身份证」，没有就用不了

获取地址：阿里云灵积平台 → https://dashscope.console.aliyun.com/
免费额度：阿里云新用户有免费调用额度，完全够用学习
配置方式：代码里是从config.py导入，你新建一个config.py文件，里面只写一行即可：
```
# config.py 文件内容
DASHSCOPE_API_KEY = "你的通义千问API-KEY"
```

3. 核心前置概念（3个，必须理解，不然看不懂代码）

① 通义千问的调用格式（固定规则）

所有大模型的调用都有固定的消息格式，通义千问也不例外，必须传一个messages列表，里面是字典，只有两种角色：

role: system：系统指令 → 告诉大模型「你是谁、要做什么、遵守什么规则」，是全局指令
role: user：用户输入 → 告诉大模型「具体的请求内容」

比如这份代码里，system指令是「你是专业翻译助手，精准翻译，不加额外解释」，user输入是「待翻译的文本」

② LangChain 1.0的核心思想：一切皆【可运行节点】

LangChain 1.0 最大的变化：把所有功能（函数、模型、解析器）都封装成「可运行的节点」，节点之间可以用 |（管道符）串联起来，形成「执行链/流水线」。

管道符 | 的作用：左边节点的运行结果，会自动作为右边节点的输入，不需要手动传参，极大简化代码
核心逻辑：输入数据 → 节点1处理 → 节点2处理 → 输出最终结果

③ 温度系数 `temperature`

代码里写了 TEMPERATURE = 0.1，这个参数是大模型生成内容的「随机度开关」，固定规则：
✅ 值越小 → 生成的内容越精准、保守、一致，适合翻译、问答、总结这类需要「准确」的场景
✅ 值越大 → 生成的内容越随机、发散、有创意，适合写文案、作诗、创作这类需要「脑洞」的场景
✅ 取值范围：0 ~ 2，翻译场景推荐 0.1~0.3 就够了。

✅ 第二步：逐段拆解代码（带图解，最核心部分）

🔶 第一段：导入依赖包

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda
import dashscope
from dashscope import Generation

# 从配置文件导入API密钥
from config import DASHSCOPE_API_KEY

StrOutputParser：字符串输出解析器 → 把大模型返回的复杂结果，最终转成「纯字符串」，方便使用
RunnableLambda：函数包装器（核心！） → 把我们自己写的「普通Python函数」，包装成 LangChain 能识别的「可运行节点」，这是连接「原生Python函数」和「LangChain执行链」的桥梁
dashscope & Generation：通义千问的SDK核心，Generation.call() 就是调用通义千问模型的入口
DASHSCOPE_API_KEY：你的通义千问密钥，用来鉴权，证明你有权调用模型

🔶 第二段：配置区（可无脑修改，最灵活）

# ==================== 配置区 ====================
# 模型配置
MODEL_NAME = "qwen-turbo"  # 使用的通义千问模型
TEMPERATURE = 0.1  # 生成温度，值越小越精确，越大越随机

# 翻译配置
SYSTEM_PROMPT_TEMPLATE = "你是专业翻译助手，将英文精准翻译成{language}，不要添加任何额外解释。"
# ==================== 配置区 ====================

# 配置通义千问 API-KEY
dashscope.api_key = DASHSCOPE_API_KEY

✅ 这里的代码全是可配置项，不用改其他地方，改这里就能实现需求变更，重点说3个点：

MODEL_NAME = "qwen-turbo"：通义千问的轻量版模型，速度快、精度够、性价比高，入门首选！还有其他可选模型：
- qwen-plus：增强版，翻译更精准，适合复杂文本
- qwen-max：旗舰版，能力最强，适合专业场景
  新手用qwen-turbo完全够用。
SYSTEM_PROMPT_TEMPLATE：翻译的核心指令模板，{language} 是占位符，后面会用真实的「目标语言」（中文/日语/法语）替换它，比如传入language=中文，最终指令就是：你是专业翻译助手，将英文精准翻译成中文，不要添加任何额外解释。
dashscope.api_key = DASHSCOPE_API_KEY：给通义千问SDK配置密钥，这一步必须有，否则调用模型会报「鉴权失败」。

🔶 第三段：核心函数1 - 同步调用函数 `call_qwen(inputs: dict)`

def call_qwen(inputs: dict) -> str:
    """
    inputs 字典必须包含两个键：language（目标语言）、text（待翻译文本）
    """
    # 拼接提示词 给大模型的指令
    system_msg = SYSTEM_PROMPT_TEMPLATE.format(language=inputs['language'])
    user_msg = inputs["text"]

    # 调用通义千问
    response = Generation.call(
        model=MODEL_NAME,
        messages=[
            {"role": "system", "content": system_msg},
            {"role": "user", "content": user_msg}
        ],
        temperature=TEMPERATURE,
        result_format="message"
    )

    # 提取并返回结果
    return response.output.choices[0].message.content

这是同步翻译的核心函数，也是你以后写LLM应用的「标准模板」，逐行解释，保证看懂：

函数入参：inputs: dict → 必须是字典，固定2个key：language（目标语言）、text（待翻译文本），比如 {"language":"中文","text":"I love LangChain"}
system_msg = ...format(...) → 把占位符替换成真实的目标语言，生成最终的系统指令
Generation.call(...) → 调用通义千问模型的核心代码，参数都是固定写法：
- model：指定用哪个千问模型
- messages：大模型的核心入参，包含系统指令+用户输入
- temperature：随机度配置
- result_format="message"：指定返回格式为「消息体」，方便后续提取内容
return response.output.choices[0].message.content → 固定的结果提取方式，通义千问返回的是一个嵌套的复杂对象，我们只需要「最终的文本内容」，记住这个写法即可！

🔶 第四段：核心函数2 - 流式调用函数 `stream_qwen(inputs: dict)`

def stream_qwen(inputs: dict):
    system_msg = SYSTEM_PROMPT_TEMPLATE.format(language=inputs['language'])
    user_msg = inputs["text"]

    responses = Generation.call(
        model=MODEL_NAME,
        messages=[
            {"role": "system", "content": system_msg},
            {"role": "user", "content": user_msg}
        ],
        stream=True,  # 流式调用的核心开关！
        temperature=TEMPERATURE
    )

    previous_content = ""  # 用于跟踪上一次的完整内容
    
    for resp in responses:
        if hasattr(resp, 'output') and hasattr(resp.output, 'text') and resp.output.text:
            current_content = resp.output.text
            
            # 计算增量内容（当前内容减去上一次的内容）
            if previous_content:
                delta = current_content[len(previous_content):]
                if delta:  # 只有当有新内容时才输出
                    yield delta
            else:
                # 第一次响应，输出完整内容
                yield current_content
            
            # 更新上一次的内容
            previous_content = current_content

这是流式翻译的核心函数，也是新手最容易看不懂的部分，重点讲清楚「流式调用的本质」和核心细节：

✅ 流式调用的本质

普通同步调用：大模型把内容「全部生成完」，一次性返回给你；
流式调用：大模型「生成一点、返回一点」，像打字一样逐步输出，前端/终端体验更好，适合长文本翻译。

✅ 核心关键点（3个）

流式开关：stream=True → 只要在Generation.call()里加这个参数，千问模型就会返回「流式生成器」，而不是一次性结果。
关键字 yield：这个函数是「生成器函数」，用yield替代return，作用是每次返回一小段新内容，而不是一次性返回全部。
增量计算逻辑：delta = current_content[len(previous_content):]
- 通义千问的流式返回是「累加的」：比如第一次返回我，第二次返回我爱，第三次返回我爱用
- 我们需要的是「每次新增的部分」：第一次我，第二次爱，第三次用
- 所以用「当前完整内容 - 上一次完整内容」，得到「新增的一小段」，这就是你看到的「打字效果」的核心逻辑！

🔶 第五段：构建 LangChain 执行链（核心灵魂，1.0写法，必掌握）

# 同步链
sync_chain = RunnableLambda(call_qwen) | StrOutputParser()
# 流式链
stream_chain = RunnableLambda(stream_qwen)

这两行代码是这份代码的精华，也是LangChain 1.0的精髓，配合图解+大白话讲透，保证你终身不忘

✅ 核心组件：`RunnableLambda(函数名)`

作用：把我们自己写的普通Python函数（call_qwen/stream_qwen），包装成 LangChain 的「可运行节点」。

普通函数 → 不能被LangChain的管道符串联，不能加入执行链
被RunnableLambda包装后 → 变成「标准节点」，可以用|串联，可以调用.invoke()/.stream()方法

✅ 管道符 `|` 的作用（⭐⭐⭐ 重中之重）

管道符 = 自动传参的流水线
左边节点的「输出结果」 → 自动作为「输入参数」传给右边节点

✅ 两个执行链的图解 & 解析

📌 同步链：`sync_chain = RunnableLambda(call_qwen) | StrOutputParser()`

输入字典 {language:xx, text:xx} 
        ↓
RunnableLambda(call_qwen) 【节点1】→ 执行同步翻译函数，返回翻译后的字符串
        ↓ （自动传参）
StrOutputParser() 【节点2】→ 把结果标准化为纯字符串（兜底处理）
        ↓
最终输出：完整的翻译文本

为什么加StrOutputParser()？做「兜底兼容」，即使大模型返回非标准格式，也能转成纯字符串，避免报错，建议所有同步链都加上。

📌 流式链：`stream_chain = RunnableLambda(stream_qwen)`

输入字典 {language:xx, text:xx} 
        ↓
RunnableLambda(stream_qwen) 【唯一节点】→ 执行流式翻译函数，返回流式生成器
        ↓
最终输出：逐段的翻译文本（打字效果）

为什么不加StrOutputParser()？流式函数本身返回的就是「字符串片段」，不需要再解析，加了反而会破坏流式效果！

🔶 第六段：本地测试代码（程序入口，直接运行）

if __name__ == "__main__":
    try:
        # ========== 测试同步调用 ==========
        print("===== 同步翻译结果 =====")
        result = sync_chain.invoke({
            "language": "中文",
            "text": "LangChain 1.0 是一个用于开发 LLM 应用的强大框架。"
        })
        print(result)

        # ========== 测试流式调用 ==========
        print("\n===== 流式翻译结果（日语）=====")
        for chunk in stream_chain.stream({
            "language": "日语",
            "text": "I love using LangChain 1.0 with Tongyi Qianwen."
        }):
            print(chunk, end="", flush=True)

    except Exception as e:
        print(f"\n运行错误：{str(e)}")

这是代码的运行入口，也是你调用翻译链的标准方式，两个核心调用方法，必须记住：

✅ 同步链调用 → `.invoke(输入字典)`

语法：sync_chain.invoke({"language":目标语言, "text":待翻译文本})
特点：一次性返回完整结果，适合短文本、需要立即拿到完整内容的场景
结果：直接赋值给变量，就能打印/保存/使用。

✅ 流式链调用 → `.stream(输入字典)`

语法：for chunk in stream_chain.stream(输入字典): print(chunk, end="", flush=True)
特点：返回一个「生成器」，需要用for循环遍历，逐段获取内容
关键参数：end="" 表示不换行，flush=True 表示立即刷新输出，这两个参数是实现「打字效果」的关键，缺一不可！

✅ 第三步：运行结果演示（你运行后能看到的效果）

同步翻译结果（中文）

===== 同步翻译结果 =====
LangChain 1.0是一款用于开发大语言模型应用的强大框架。

流式翻译结果（日语）

===== 流式翻译结果（日语）=====
私はLangChain 1.0を通義千問と組み合わせて使うのが大好きです。

流式效果在终端里是「一个字一个字慢慢出现」，这里文字展示不出来，你运行代码就能直观看到！

✅ 第四步：怎么修改代码，实现你的需求？（举一反三，重中之重）

学会改代码才是真的学会！这份代码是「模板级」的，只需要改【配置区】和【入参】，就能实现各种翻译需求，不用改核心逻辑，给你几个最常用的修改案例，直接复制用：

✅ 案例1：把「英译中」改成「中译英」

只改配置区的系统指令模板即可：

SYSTEM_PROMPT_TEMPLATE = "你是专业翻译助手，将中文精准翻译成{language}，不要添加任何额外解释。"

调用：

result = sync_chain.invoke({
    "language": "英文",
    "text": "我喜欢用LangChain结合通义千问做开发。"
})

✅ 案例2：支持「多语言互译」（比如中译法、英译韩）

指令模板不用改！只需要改调用时的language值，千问模型能支持所有主流语言：

# 中译法
result = sync_chain.invoke({"language": "法语", "text": "人工智能改变世界"})
# 英译韩
for chunk in stream_chain.stream({"language": "韩语", "text": "LLM is the future of AI"}):
    print(chunk, end="", flush=True)

✅ 案例3：提高翻译的「文采」（比如文学翻译）

只需要改系统指令模板，加需求描述即可：

SYSTEM_PROMPT_TEMPLATE = "你是专业文学翻译助手，将英文精准翻译成{language}，翻译结果优美流畅，符合中文表达习惯，不要添加任何额外解释。"

✅ 案例4：翻译时「保留原文格式」（比如带标点、换行）

SYSTEM_PROMPT_TEMPLATE = "你是专业翻译助手，将英文精准翻译成{language}，严格保留原文的标点和换行格式，不要添加任何额外解释。"

✅ 第五步：总结（所有核心知识点+必背）

这份代码虽然短，但包含了LangChain 1.0开发的所有核心基础知识点，学会这些，你就能写80%的LLM入门应用，我帮你整理成「必背清单」，看完直接记牢：

✅ 核心组件必背

RunnableLambda：包装普通函数 → LangChain可运行节点
StrOutputParser：标准化输出 → 纯字符串
管道符 |：自动传参，串联节点，形成执行链

✅ 调用方法必背

同步调用：chain.invoke(输入数据) → 一次性返回结果
流式调用：for chunk in chain.stream(输入数据): → 逐段返回结果

✅ 通义千问必背

调用入口：Generation.call()
流式开关：stream=True
结果提取：response.output.choices[0].message.content
温度系数：越小越精准，越大越随机

✅ 流式核心必背

流式函数用 yield 替代 return
增量计算：current_content[len(previous_content):]
打印流式效果：print(chunk, end="", flush=True)

✅ 最后：运行报错解决方案（新手必看，避坑）

如果运行代码报错，99%是以下3个原因，按顺序排查即可：

鉴权失败：Invalid api key → 检查你的DASHSCOPE_API_KEY是否正确，是否在config.py里写对了
依赖缺失：ModuleNotFoundError → 重新执行pip install langchain-core dashscope
入参错误：KeyError: 'language' → 调用时的字典必须包含language和text两个key，不能少

【完整代码】

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda
import dashscope
from dashscope import Generation

# 从配置文件导入API密钥
from config import DASHSCOPE_API_KEY

# ==================== 配置区 ====================
# 模型配置
MODEL_NAME = "qwen-turbo"  # 使用的通义千问模型
TEMPERATURE = 0.1  # 生成温度，值越小越精确，越大越随机

# 翻译配置
SYSTEM_PROMPT_TEMPLATE = "你是专业翻译助手，将英文精准翻译成{language}，不要添加任何额外解释。"
# ==================== 配置区 ====================

# 配置通义千问 API-KEY
dashscope.api_key = DASHSCOPE_API_KEY


# 2. 定义原生调用函数（直接接收字典输入）
def call_qwen(inputs: dict) -> str:
    """
    inputs 字典必须包含两个键：language（目标语言）、text（待翻译文本）
    """
    # 拼接提示词 给大模型的指令
    system_msg = SYSTEM_PROMPT_TEMPLATE.format(language=inputs['language'])
    user_msg = inputs["text"]

    # 调用通义千问
    response = Generation.call(
        model=MODEL_NAME,
        messages=[
            {"role": "system", "content": system_msg},
            {"role": "user", "content": user_msg}
        ],
        temperature=TEMPERATURE,  # 精度的调控 越小精度越高 越大越随机
        result_format="message"
    )

    # 提取并返回结果
    return response.output.choices[0].message.content


# 3. 定义流式调用函数（可选）
def stream_qwen(inputs: dict):
    # 拼接提示词 给大模型的指令
    system_msg = SYSTEM_PROMPT_TEMPLATE.format(language=inputs['language'])
    user_msg = inputs["text"]

    responses = Generation.call(
        model=MODEL_NAME,
        messages=[
            {"role": "system", "content": system_msg},
            {"role": "user", "content": user_msg}
        ],
        stream=True,
        temperature=TEMPERATURE
    )

    previous_content = ""  # 用于跟踪上一次的完整内容
    
    for resp in responses:
        if hasattr(resp, 'output') and hasattr(resp.output, 'text') and resp.output.text:
            current_content = resp.output.text
            
            # 计算增量内容（当前内容减去上一次的内容）
            if previous_content:
                delta = current_content[len(previous_content):]
                if delta:  # 只有当有新内容时才输出
                    yield delta
            else:
                # 第一次响应，输出完整内容
                yield current_content
            
            # 更新上一次的内容
            previous_content = current_content


# 4. 构建 LangChain 链（直接使用自定义函数）sync_chain与stream_chain是已经封装好的翻译工具
# 同步链 |称之为管道符 作用是把左边的执行结果，自动传给右边作为输入 RunnableLambda是将江汉包装成为Langchain的可执行节点
sync_chain = RunnableLambda(call_qwen) | StrOutputParser()
# 流式链 效果就是一个字一个的翻译出来
stream_chain = RunnableLambda(stream_qwen)

# 5. 本地测试
if __name__ == "__main__":
    try:
        # ========== 测试同步调用 ==========
        print("===== 同步翻译结果 =====")
        result = sync_chain.invoke({
            "language": "中文",
            "text": "LangChain 1.0 是一个用于开发 LLM 应用的强大框架。"
        })
        print(result)

        # ========== 测试流式调用 ==========
        print("\n===== 流式翻译结果（日语）=====")
        for chunk in stream_chain.stream({
            "language": "日语",
            "text": "I love using LangChain 1.0 with Tongyi Qianwen."
        }):print(chunk, end="", flush=True)

    except Exception as e:
        print(f"\n运行错误：{str(e)}")

2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

8个降AI率工具推荐！本科生高效避坑指南

2048 AI社区

聊天就能生成RPA自动化流程，这款工作流软件比n8n更好用？

就在前不久，我参观了一家专门做直播获客卖车的企业，它们有一个流程设计吸引到我，通过直播留资来的意向用户会被自动拉进企微群，群里有专门的客服机器人发送车型信息、咨询，回复购车政策等问题，并定时总结客户信息回传到数据库，7*24小时不停业，这中间完全不需要人来参与，极大的提升了运营效率。现在普遍降本增效的大环境下，RPA绝对是企业和个人自媒体提升运营效率的一大利器，而且随着八爪鱼RPA AI生成功能的

2048 AI社区

AI率从81.2%降至7.9%，这款降重工具真的好用吗？

2048 AI社区

所有评论(0)

查看更多评论

稀稀落落：：

@weixin_64168302

已为社区贡献2条内容

【001】LangChain + 通义千问大模型实现多语言互译-可运行(完整代码在结尾）

稀稀落落：：

零基础吃透这份 LangChain+通义千问 翻译代码（全程大白话+图解+实操）

✅ 先搞懂【整体功能】

✅ 第一步：环境准备 & 前置知识（必须先做，否则运行报错）

1. 安装依赖包

2. 通义千问API-KEY准备

3. 核心前置概念（3个，必须理解，不然看不懂代码）

① 通义千问的调用格式（固定规则）

② LangChain 1.0的核心思想：一切皆【可运行节点】

③ 温度系数 temperature

✅ 第二步：逐段拆解代码（带图解，最核心部分）

🔶 第一段：导入依赖包

🔶 第二段：配置区（可无脑修改，最灵活）

🔶 第三段：核心函数1 - 同步调用函数 call_qwen(inputs: dict)

🔶 第四段：核心函数2 - 流式调用函数 stream_qwen(inputs: dict)

✅ 流式调用的本质

✅ 核心关键点（3个）

🔶 第五段：构建 LangChain 执行链（核心灵魂，1.0写法，必掌握）

✅ 核心组件：RunnableLambda(函数名)

✅ 管道符 | 的作用（⭐⭐⭐ 重中之重）

✅ 两个执行链的图解 & 解析

📌 同步链：sync_chain = RunnableLambda(call_qwen) | StrOutputParser()

📌 流式链：stream_chain = RunnableLambda(stream_qwen)

🔶 第六段：本地测试代码（程序入口，直接运行）

✅ 同步链调用 → .invoke(输入字典)

✅ 流式链调用 → .stream(输入字典)

✅ 第三步：运行结果演示（你运行后能看到的效果）

同步翻译结果（中文）

流式翻译结果（日语）

✅ 第四步：怎么修改代码，实现你的需求？（举一反三，重中之重）

✅ 案例1：把「英译中」改成「中译英」

✅ 案例2：支持「多语言互译」（比如中译法、英译韩）

✅ 案例3：提高翻译的「文采」（比如文学翻译）

✅ 案例4：翻译时「保留原文格式」（比如带标点、换行）

✅ 第五步：总结（所有核心知识点+必背）

✅ 核心组件必背

✅ 调用方法必背

✅ 通义千问必背

✅ 流式核心必背

✅ 最后：运行报错解决方案（新手必看，避坑）

【完整代码】

所有评论(0)

稀稀落落：：

零基础吃透这份 LangChain+通义千问翻译代码（全程大白话+图解+实操）

③ 温度系数 `temperature`

🔶 第三段：核心函数1 - 同步调用函数 `call_qwen(inputs: dict)`

🔶 第四段：核心函数2 - 流式调用函数 `stream_qwen(inputs: dict)`

✅ 核心组件：`RunnableLambda(函数名)`

✅ 管道符 `|` 的作用（⭐⭐⭐ 重中之重）

📌 同步链：`sync_chain = RunnableLambda(call_qwen) | StrOutputParser()`

📌 流式链：`stream_chain = RunnableLambda(stream_qwen)`

✅ 同步链调用 → `.invoke(输入字典)`

✅ 流式链调用 → `.stream(输入字典)`