构建LLM应用的框架
问题1：代码重复，难以维护
问题2：模型切换成本高
❌ 原来用OpenAI的代码：
from openai import OpenAI
client = OpenAI(api_key="...")
response = client.chat.completions.create(...)
❌ 想换成通义千问，要全部重写：
import dashscope
response = dashscope.Generation.call(...)
结果：换个模型，整个项目都要改！
问题3：复杂功能难实现
1. 统一接口，轻松切换模型
✅ 用OpenAI：
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo")
✅ 换成通义千问，只改一行：
from langchain_community.llms import Tongyi
llm = Tongyi(model="qwen-turbo")
✅ 本地用Ollama（不用Key，适合离线/内网）：
from langchain_community.chat_models import ChatOllama
llm = ChatOllama(model="llama3.1", base_url="http://localhost:11434")
切换逻辑不变：你后续的 Prompt / Chain / Memory / RAG 代码仍然可以沿用。
其他代码完全不用改！
2. 封装常用功能，开箱即用
3. 模块化设计，灵活组合
像搭积木一样，把不同组件组合起来，快速构建复杂应用。
LangChain支持几十种大模型，覆盖国内外主流厂商！
• GPT-4, GPT-4 Turbo
• GPT-3.5 Turbo
• GPT-4o
• Claude 3 Opus
• Claude 3 Sonnet
• Claude 3 Haiku
• Gemini Pro
• Gemini Ultra
• PaLM 2
• Llama 2
• Llama 3
• Code Llama
• Qwen-Turbo
• Qwen-Plus
• Qwen-Max
• ERNIE-Bot
• ERNIE-Bot-turbo
• ERNIE-Bot-4
• GLM-4
• GLM-3-Turbo
• ChatGLM-Pro
• Spark-3.5
• Spark-3.0
• Spark-Lite
• Moonshot-v1-8k
• Moonshot-v1-32k
• Moonshot-v1-128k
• Doubao-pro
• Doubao-lite
支持多种场景
核心优势：只需要改一行代码，其他业务逻辑完全不变！
# 安装LangChain核心库
pip install langchain
# 安装通义千问支持
pip install langchain-community
pip install dashscope
# 安装传统 Chain API（包含 OutputFixingParser 等）
pip install langchain-classicfrom langchain_community.llms import Tongyi
import os
# 设置API Key（从环境变量获取）
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")
# 初始化通义千问模型
llm = Tongyi(
model_name="qwen-turbo", # 模型名称
temperature=0.7, # 温度参数
top_p=0.8, # top_p采样
max_tokens=2000 # 最大生成长度
)
# 直接调用
response = llm.invoke("介绍一下Python语言")
print(response)同样写法，换成 OpenAI：
from langchain_openai import ChatOpenAI
import os
# 设置 API Key（建议用环境变量）
os.environ["OPENAI_API_KEY"] = "your-api-key"
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
response = llm.invoke("介绍一下Python语言")
print(response)同样写法，本地用 Ollama：
from langchain_community.chat_models import ChatOllama
# 先确保本机已启动：ollama serve
# 并拉取模型：ollama pull llama3.1
llm = ChatOllama(model="llama3.1", base_url="http://localhost:11434", temperature=0.7)
response = llm.invoke("介绍一下Python语言")
print(response)同样写法，换成 豆包（火山方舟 / OpenAI 兼容接口）：
from langchain_openai import ChatOpenAI
import os
# 初始化豆包模型（兼容OpenAI接口）
llm = ChatOpenAI(
model="doubao-seed-1-8-251228", # 替换为具体豆包模型ID（见下方推荐）
openai_api_key=os.environ.get("VOLC_API_KEY"), # 从火山方舟控制台获取
openai_api_base="https://ark.cn-beijing.volces.com/api/v3", # 火山方舟API端点
max_tokens=2048,
temperature=0.7,
streaming=True
)
# 发送请求
for chunk in llm.stream("用通俗的话解释一下：什么是LangChain？"):
text = getattr(chunk, "content", chunk)
if isinstance(text, list):
text = "".join(
item.get("text", "")
for item in text
if isinstance(item, dict) and item.get("type") == "text"
)
if text:
print(text, end="", flush=True)
print()Python是一种高级编程语言，由Guido van Rossum于1991年创建。
它以简洁、易读的语法著称，适合初学者学习，同时也被广泛应用于
Web开发、数据分析、人工智能、科学计算等领域...流式输出的体验更像“打字机”，适合做聊天 UI 或者长文本生成（用户不用傻等）。
from langchain_community.llms import Tongyi
import os
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")
# 开启 streaming=True
llm = Tongyi(model_name="qwen-turbo", temperature=0.7, streaming=True)
# 逐段输出（像打字一样）
for chunk in llm.stream("用通俗的话解释一下：什么是LangChain？"):
print(chunk, end="", flush=True)
print()LangChain 是一个用来快速搭建大模型应用的开发框架，它把“提示词、记忆、工具、检索、链路编排”等常见能力封装好，让你像搭积木一样组合出应用。（逐字输出）Token是大模型处理文本的基本单位，理解Token对于控制成本和长度非常重要。
Token不等于字符！大模型会把文本切分成Token，规则如下：
使用tiktoken库计算文本的Token数量。
pip install tiktoken"""
Token计数示例 - 了解文本消耗多少Token
"""
import tiktoken
# 获取tokenizer（使用cl100k_base作为近似估算）
encoding = tiktoken.get_encoding("cl100k_base")
# ========== 示例1：中文文本 ==========
text_cn = "黄河之水天上来"
tokens_cn = encoding.encode(text_cn)
print(f"中文文本: {text_cn}")
print(f"Token编码: {tokens_cn}")
print(f"Token数量: {len(tokens_cn)}")
print()
# ========== 示例2：英文文本 ==========
text_en = "Hello, how are you?"
tokens_en = encoding.encode(text_en)
print(f"英文文本: {text_en}")
print(f"Token编码: {tokens_en}")
print(f"Token数量: {len(tokens_en)}")
print()
# ========== 示例3：混合文本 ==========
text_mix = "我喜欢Python编程，it's amazing!"
tokens_mix = encoding.encode(text_mix)
print(f"混合文本: {text_mix}")
print(f"Token数量: {len(tokens_mix)}")
print()
# ========== 示例4：计算对话的Token数 ==========
messages = [
{"role": "system", "content": "你是一个Python助手"},
{"role": "user", "content": "什么是列表？"},
{"role": "assistant", "content": "列表是Python中的一种数据结构..."}
]
total_tokens = 0
for msg in messages:
content = msg["content"]
tokens = encoding.encode(content)
total_tokens += len(tokens)
print(f"{msg['role']}: {len(tokens)} tokens")
print(f"\n总Token数: {total_tokens}")中文文本: 黄河之水天上来
Token编码: [104595, 103, 233, 99, 105, 33890, 99, 244, 73138, 99, 101, 17792]
Token数量: 12
👉 7个汉字 = 12个Token（平均1.7个Token/字）
英文文本: Hello, how are you?
Token编码: [9906, 11, 1268, 527, 499, 30]
Token数量: 6
👉 4个单词 = 6个Token（包含标点）
混合文本: 我喜欢Python编程，it's amazing!
Token数量: 15
对话Token统计:
system: 11 tokens
user: 8 tokens
assistant: 18 tokens
总Token数: 37调用API后，可以从响应中获取实际消耗的Token数量。
# 建议新建虚拟环境后安装
pip install -U langchain langchain-community langchain-openai
pip install -U dashscope tiktoken
注意：很多 LangChain 的对话模型（Chat Model）会把 Token 使用量放在返回的 AIMessage 元数据里（常见字段如 usage_metadata / response_metadata / token_usage）。
get_openai_callback 主要用于 OpenAI（或兼容 OpenAI 的接口）的 Token/成本统计，这里不展开。
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage
import os
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")
chat_model = ChatTongyi(model_name="qwen-turbo", temperature=0)
response = chat_model.invoke([HumanMessage(content="介绍一下Python语言")])
print(f"回复: {response.content}")
# 尝试从返回消息中读取 token usage（不同版本/不同模型字段名可能略有差异）
usage = None
for key in ("usage_metadata", "response_metadata", "token_usage"):
value = getattr(response, key, None)
if value:
usage = value
break
print(f"\n--- Token统计（从AIMessage元数据读取） ---")
print(usage)回复: Python是一种高级编程语言...
--- Token统计 ---
{'input_tokens': 8, 'output_tokens': 156, 'total_tokens': 164}用同一个问题，测试不同temperature的效果差异。
from langchain_community.llms import Tongyi
import os
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")
# 测试问题
question = "用一句话描述春天"
# 实验1：低温度（精确、确定）
print("=== Temperature = 0.1 ===")
llm_low = Tongyi(model_name="qwen-turbo", temperature=0.1)
for i in range(3):
print(f"第{i+1}次: {llm_low.invoke(question)}")
# 实验2：中温度（平衡）
print("\n=== Temperature = 0.7 ===")
llm_mid = Tongyi(model_name="qwen-turbo", temperature=0.7)
for i in range(3):
print(f"第{i+1}次: {llm_mid.invoke(question)}")
# 实验3：高温度（创意、随机）
print("\n=== Temperature = 1.5 ===")
llm_high = Tongyi(model_name="qwen-turbo", temperature=1.5)
for i in range(3):
print(f"第{i+1}次: {llm_high.invoke(question)}")=== Temperature = 0.1 ===
第1次: 春天是万物复苏、生机勃勃的季节。
第2次: 春天是万物复苏、生机勃勃的季节。
第3次: 春天是万物复苏、生机勃勃的季节。
👉 结论：低温度输出非常稳定，几乎一模一样
=== Temperature = 0.7 ===
第1次: 春天是万物复苏、花开遍野的美好季节。
第2次: 春天是大地回春、生机盎然的时节。
第3次: 春天是温暖和煦、充满希望的季节。
👉 结论：中温度有一定变化，但都合理
=== Temperature = 1.5 ===
第1次: 春天像一位画家，用绿色和鲜花装点大地。
第2次: 春风拂面，万物苏醒，这就是春天的魔力！
第3次: 春天？那是冰雪消融后的第一声鸟鸣。
👉 结论：高温度输出很有创意，但可能不够稳定LangChain的核心是模块化设计，像搭积木一样组合不同组件。
模型是整个应用的核心，负责理解和生成文本。
from langchain_community.llms import Tongyi
from langchain_community.chat_models import ChatTongyi
# 文本生成模型（适合单次生成）
llm = Tongyi(model_name="qwen-turbo")
response = llm.invoke("什么是Python？")
# 对话模型（支持多轮对话）
chat_model = ChatTongyi(model_name="qwen-turbo")
messages = [
("system", "你是Python专家"),
("user", "什么是列表？")
]
response = chat_model.invoke(messages)
print(response.content)
在“对话模型”里，LangChain 会把一轮对话表示成一个 messages 列表。
你前面看到的 ("system", "...") / ("user", "...") 其实是简写；更“显式 / 更可控”的写法是直接构造消息对象：
SystemMessage、HumanMessage、AIMessage。
用来放系统级指令：角色设定、风格约束、安全边界、输出格式偏好等。 通常放在对话最前面，并且在多轮对话中会被反复带上。
表示用户输入（你对模型说的话）。
绝大多数场景只需要传 content 文本；复杂场景也可以携带额外元数据。
表示模型输出。
你调用 chat_model.invoke(...) 得到的返回值通常就是一个 AIMessage；常用字段是 .content（文本内容）。
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
chat_model = ChatTongyi(model_name="qwen-turbo")
messages = [
SystemMessage(content="你是Python老师，用通俗中文回答，尽量给出例子"),
HumanMessage(content="解释一下什么是列表（list）"),
]
result = chat_model.invoke(messages)
print(type(result)) # <class 'langchain_core.messages.ai.AIMessage'>
print(result.content) # 模型输出文本
# 如果你要把历史对话继续带上去，也可以显式追加 AIMessage
messages.append(AIMessage(content=result.content))
messages.append(HumanMessage(content="再给一个包含 for 循环的例子"))
result2 = chat_model.invoke(messages)
print(result2.content)提示模板让你可以复用和参数化提示词，避免重复编写。
pip install -U langchain-core langchainfrom langchain_core.prompts import PromptTemplate, ChatPromptTemplate
# 简单提示模板
prompt = PromptTemplate.from_template("将{text}翻译成{language}")
formatted = prompt.format(text="你好", language="英文")
print(formatted) # 输出：将你好翻译成英文
# 对话提示模板
chat_prompt = ChatPromptTemplate.from_messages([
("system", "你是{role}"),
("user", "{input}")
])
messages = chat_prompt.format_messages(
role="Python老师",
input="解释什么是变量"
)
print(messages)输出解析器将模型的原始输出转换成我们需要的格式。
问题：模型返回的是复杂对象
# 不使用Parser
result = llm.invoke("介绍Python")
print(result)
# 输出：AIMessage(content='Python是...', response_metadata={...})
# 👉 返回的是对象，不是纯文本！解决：使用StrOutputParser提取文本
# 使用Parser
from langchain_core.output_parsers import StrOutputParser
from langchain_community.llms import Tongyi
# 初始化LLM
llm = Tongyi(model_name="qwen-turbo")
# 创建链
chain = llm | StrOutputParser()
result = chain.invoke("介绍Python")
print(result)
# 输出：Python是一种高级编程语言...
# 👉 直接返回纯文本字符串！你可以把 Parser 理解成“最后一步的格式化器”：把模型输出从字符串/消息对象，变成你工程里真正要用的字符串 / dict / list / 结构化对象。 下面是工程里最常见的一组 Parser（按使用频率从高到低排列）：
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser, CommaSeparatedListOutputParser
from langchain_community.llms import Tongyi
import os
llm = Tongyi(model_name="qwen-turbo")
# ========== 1. StrOutputParser - 字符串解析器 ==========
# 最常用！将输出转换为纯字符串
str_parser = StrOutputParser()
chain = llm | str_parser
result = chain.invoke("用一句话介绍Python")
print(f"类型: {type(result)}") #
print(f"内容: {result}")
# ========== 2. JsonOutputParser - JSON解析器 ==========
# 将输出解析为JSON对象
from langchain_core.prompts import PromptTemplate
json_parser = JsonOutputParser()
prompt = PromptTemplate.from_template(
"""请将以下用户信息提取为JSON格式：\n{user_input}\n\n{format_instructions}"""
)
chain = prompt | llm | json_parser
result = chain.invoke({
"user_input": "介绍一下李明，30岁，程序员",
"format_instructions": json_parser.get_format_instructions()
})
print(f"类型: {type(result)}") #
print(f"内容: {result}") # {'姓名': '李明', '年龄': 30, '职业': '程序员'}
# ========== 3. CommaSeparatedListOutputParser - 逗号分隔列表解析器 ==========
# 将输出解析为列表
list_parser = CommaSeparatedListOutputParser()
prompt = PromptTemplate.from_template(
"""请列出3种编程语言，用逗号分隔：\n{format_instructions}"""
)
chain = prompt | llm | list_parser
result = chain.invoke({"format_instructions": list_parser.get_format_instructions()})
print(f"类型: {type(result)}") #
print(f"内容: {result}") # ['Python', 'Java', 'JavaScript']
# ========== 4. PydanticOutputParser - Pydantic结构化解析器 ==========
# 把输出解析为“强类型对象”（字段缺失/类型不对会报错）
from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser
class UserInfo(BaseModel):
name: str = Field(description="姓名")
age: int = Field(description="年龄")
job: str = Field(description="职业")
pydantic_parser = PydanticOutputParser(pydantic_object=UserInfo)
pyd_prompt = PromptTemplate.from_template(
"""从文本中提取用户信息，并按指定 JSON 格式输出：\n{text}\n\n{format_instructions}"""
)
pyd_chain = pyd_prompt | llm | pydantic_parser
pyd_result = pyd_chain.invoke({
"text": "李明，30岁，程序员",
"format_instructions": pydantic_parser.get_format_instructions(),
})
print(f"类型: {type(pyd_result)}")
print(f"内容: {pyd_result}")
# ========== 5. OutputFixingParser - 解析失败自动修正（真实使用） ==========
# 真实使用 OutputFixingParser：当解析失败时，自动让模型修正后重试
from langchain_classic.output_parsers import OutputFixingParser
# 创建一个基础 parser（这里用 JsonOutputParser）
base_parser = JsonOutputParser()
# 创建 OutputFixingParser，传入 LLM 和基础 parser
fixing_parser = OutputFixingParser.from_llm(
llm=llm,
parser=base_parser
)
# 测试：故意给一个错误的 JSON
bad_json = "{'name': 'Li Ming', 'age': 30, 'job': 'Engineer'}" # 单引号，不是合法 JSON
try:
fixed_demo = fixing_parser.parse(bad_json)
print(f"✅ 修复成功！类型: {type(fixed_demo)}")
print(f"内容: {fixed_demo}")
except Exception as e:
print(f"❌ 修复失败: {e}")
# ========== 6. RetryWithErrorOutputParser - 带错误原因的重试（真实使用） ==========
# 真实使用 RetryWithErrorOutputParser：解析失败时把错误原因喂回模型重试
from langchain_classic.output_parsers import RetryWithErrorOutputParser
# 创建 RetryWithErrorOutputParser
retry_parser = RetryWithErrorOutputParser.from_llm(
llm=llm,
parser=base_parser
)
# 测试：给一个完全不是 JSON 的格式
raw_model_output = "姓名=李明; 年龄=三十; 职业=程序员" # 非 JSON
try:
retry_demo = retry_parser.parse(raw_model_output)
print(f"✅ 重试成功！类型: {type(retry_demo)}")
print(f"内容: {retry_demo}")
except Exception as e:
print(f"❌ 重试失败: {e}")# StrOutputParser
类型:
内容: Python是一种高级编程语言，以简洁易读著称。
# JsonOutputParser
类型:
内容: {'姓名': '李明', '年龄': 30, '职业': '程序员'}
# ListOutputParser
类型:
内容: ['Python', 'Java', 'JavaScript']
链（Chain）把多个组件按“数据流”串联起来，实现可复用的工作流。在 LCEL 里，Prompt / Model / OutputParser 本质上都是 Runnable（可组合的“函数”）：输入一个对象（字符串或 dict），输出另一个对象（字符串/字典/列表）。
你最常见会用到的链路形态是：Prompt → LLM → Parser；进阶一点是并行组合（多条子链同时跑）和路由分支（先识别意图再分发）。
工程上，Chains 的价值在于：可测试（每段可单独 invoke）、可复用（prompt/解析器/子链可抽出复用）、可观测（链路清晰便于加回调/日志）、可扩展（后面做 RAG/Agent 也基本是在“拼链”）。
手写 Prompt + 调用模型（更直观）
text = prompt.format(... )
result = llm.invoke(text)
使用 | 管道操作符
chain = prompt | llm
更简洁、更直观！
chain = prompt | llm | parser 之后，选择用哪种方法调用，取决于你的业务形态：
asyncio.gather 能同时发起多个调用。
invoke = 单次同步；ainvoke = 单次异步（可并发）；batch = 多输入批量；stream = 流式逐步输出。
from langchain_community.llms import Tongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os
import asyncio
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")
llm = Tongyi(model_name="qwen-turbo")
prompt = PromptTemplate.from_template("用一句话介绍{topic}")
chain = prompt | llm | StrOutputParser()
# 1) invoke：同步
print(chain.invoke({"topic": "人工智能"}))
# 2) batch：批量
print(chain.batch([
{"topic": "人工智能"},
{"topic": "区块链"},
]))
# 3) stream：流式
for chunk in chain.stream({"topic": "机器学习"}):
print(chunk, end="", flush=True)
print()
# 4) ainvoke：异步 + 并发
async def main():
tasks = [
chain.ainvoke({"topic": "人工智能"}),
chain.ainvoke({"topic": "量子计算"}),
chain.ainvoke({"topic": "大语言模型"}),
]
results = await asyncio.gather(*tasks)
for i, r in enumerate(results, 1):
print(i, r)
asyncio.run(main())pip install -U langchain-core langchain-community dashscopefrom langchain_community.llms import Tongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")
llm = Tongyi(model_name="qwen-turbo")
prompt = PromptTemplate.from_template("用一句话介绍{topic}")
# ========== 方式对比：传统 vs LCEL ==========
# 传统方式（手写 prompt + 调用模型）
text = prompt.format(topic="人工智能")
result = llm.invoke(text)
print(result)
# LCEL方式（推荐）⭐
chain_new = prompt | llm | StrOutputParser()
result = chain_new.invoke({"topic": "人工智能"})
print(result) # 直接返回字符串，更简洁！
# ========== LCEL完整示例 ==========
# 创建链：prompt | model | output_parser
chain = prompt | llm | StrOutputParser()
# 方法1：invoke() - 单次调用
result = chain.invoke({"topic": "人工智能"})
print(result)
# 输出：人工智能是模拟人类智能的技术...
# ========== 方法2：batch() - 批量处理 ==========
# 一次处理多个输入
results = chain.batch([
{"topic": "人工智能"},
{"topic": "区块链"},
{"topic": "量子计算"}
])
for i, result in enumerate(results, 1):
print(f"{i}. {result}")
# ========== 方法3：stream() - 流式输出 ==========
# 像打字一样逐字显示
for chunk in chain.stream({"topic": "机器学习"}):
print(chunk, end="", flush=True)
# ========== LCEL顺序链示例 ==========
# 使用 | 操作符串联多个步骤
prompt1 = PromptTemplate.from_template("为{topic}写一个故事大纲")
prompt2 = PromptTemplate.from_template("根据以下大纲，写一个详细的故事：\n{outline}")
# 方式1：分步创建
outline_chain = prompt1 | llm | StrOutputParser()
story_chain = prompt2 | llm | StrOutputParser()
# 先生成大纲
outline = outline_chain.invoke({"topic": "太空探险"})
print(f"大纲：{outline}")
# 再生成故事
story = story_chain.invoke({"outline": outline})
print(f"故事：{story}")
# 方式2：使用RunnablePassthrough自动传递
from langchain_core.runnables import RunnablePassthrough
# 完整的链式调用
full_chain = (
{"outline": prompt1 | llm | StrOutputParser()} # 生成大纲
| prompt2 # 使用大纲生成故事
| llm
| StrOutputParser()
)
story = full_chain.invoke({"topic": "太空探险"})
print(story)# 传统方式 vs LCEL方式对比
传统方式输出：{'text': '人工智能是模拟人类智能的技术...'}
LCEL方式输出：人工智能是模拟人类智能的技术...
👉 LCEL更简洁，直接返回字符串！
# invoke() - 单次调用
人工智能是模拟人类智能的技术，包括学习、推理和自我修正能力。
# batch() - 批量处理
1. 人工智能是模拟人类智能的技术...
2. 区块链是一种分布式账本技术...
3. 量子计算利用量子力学原理进行计算...
# stream() - 流式输出
机器学习是人工智能的一个分支，通过数据训练模型...（逐字显示）
# LCEL顺序链
大纲：故事大纲：主角是宇航员张明，在执行任务时发现神秘星球...
故事：在遥远的未来，宇航员张明驾驶飞船探索未知星系...| 操作符，代码更易读推荐：新项目优先使用LCEL方式！
这是“链”最常见的实战形态之一：
先用一个小 Prompt 识别用户意图（翻译/总结/解释代码），再把请求分发到不同子链。
优点是：入口统一、新增能力只需要加一个分支，不会把主逻辑写成一堆 if-else。
prompt | model | parser 的底层形态）。
assign 增补字段。
intent、trace_id、validated_text 写回上下文）。
pip install -U langchain-core langchain-community dashscopefrom langchain_community.llms import Tongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch, RunnableLambda
import os
# 从环境变量读取 API Key（请确保已设置 DASHSCOPE_API_KEY）
os.environ["DASHSCOPE_API_KEY"] = os.environ.get("DASHSCOPE_API_KEY", "")
llm = Tongyi(model_name="qwen-turbo", temperature=0)
parser = StrOutputParser()
# 1) Router：识别意图，只输出一个标签：translate / summarize / explain
router_prompt = PromptTemplate.from_template(
"""你是一个路由器，只输出一个标签：translate / summarize / explain。
用户输入：{text}
判断规则：
- 如果用户要求翻译（含“翻译/英文/中文/日文”等），输出 translate
- 如果用户要求总结/摘要/要点，输出 summarize
- 如果用户给了代码或要求解释代码，输出 explain
只输出标签，不要输出其他内容。"""
)
router_chain = router_prompt | llm | parser
# 2) 三个子链
translate_prompt = PromptTemplate.from_template("把下面内容翻译成英文：\n{text}\n\n英文：")
summarize_prompt = PromptTemplate.from_template("把下面内容总结成 3 条要点：\n{text}\n\n要点：")
explain_prompt = PromptTemplate.from_template(
"""你是资深开发者。请解释下面代码的作用、关键步骤，并指出一个可改进点：
{text}
"""
)
translate_chain = translate_prompt | llm | parser
summarize_chain = summarize_prompt | llm | parser
explain_chain = explain_prompt | llm | parser
# 3) 路由 + 分支执行
def pick_intent(x: dict) -> str:
intent = router_chain.invoke({"text": x["text"]}).strip().lower()
return intent
intent_runnable = RunnableLambda(pick_intent)
routed_chain = (
{"intent": intent_runnable, "text": RunnableLambda(lambda x: x["text"])}
| RunnableBranch(
(lambda x: x["intent"] == "translate", translate_chain),
(lambda x: x["intent"] == "summarize", summarize_chain),
(lambda x: x["intent"] == "explain", explain_chain),
explain_chain,
)
)
print("=== 1) 翻译 ===")
print(routed_chain.invoke({"text": "把这句话翻译成英文：我正在学习 LangChain"}))
print("\n=== 2) 总结 ===")
print(routed_chain.invoke({"text": "请把下面这段话总结一下：LangChain 是一个用来构建大模型应用的框架，它提供了提示词、链、记忆、检索等组件。"}))
print("\n=== 3) 解释代码 ===")
print(routed_chain.invoke({"text": "def add(a, b):\n return a + b"}))=== 1) 翻译 ===
I am learning LangChain.
=== 2) 总结 ===
1. LangChain 是用于构建大模型应用的开发框架。
2. 它提供提示词、链、记忆、检索等可组合组件。
3. 通过组件组合可以快速搭建复杂的 LLM 应用。
=== 3) 解释代码 ===
这段代码定义了一个 add 函数，用于返回两个参数的和。
关键步骤是函数定义与直接返回表达式 a + b。
可改进点：增加类型注解与输入校验，例如 def add(a: int, b: int) -> int。
Model + Prompt + Chain = 完整应用
就像：大脑 + 指令 + 流程 = 智能助手
使用Prompt模板和通义千问构建翻译助手。
from langchain_community.llms import Tongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os
# 初始化模型（翻译任务用低温度）
llm = Tongyi(
model_name="qwen-turbo",
temperature=0.1, # 低温度，保证翻译准确
top_p=0.8
)
# 创建提示模板
template = """你是一个专业的翻译助手。
请将以下{source_lang}文本翻译成{target_lang}，要求：
1. 准确传达原文含义
2. 符合目标语言习惯
3. 保持专业术语的准确性
原文：{text}
翻译："""
prompt = PromptTemplate(
input_variables=["source_lang", "target_lang", "text"],
template=template
)
# 使用新的LCEL语法创建链（替代已弃用的LLMChain）
translation_chain = prompt | llm | StrOutputParser()
# 使用示例
result = translation_chain.invoke({
"source_lang": "中文",
"target_lang": "英文",
"text": "人工智能正在改变我们的生活方式"
})
print(result)Artificial intelligence is transforming our way of life.让AI解释代码的功能和逻辑。
from langchain_community.llms import Tongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os
from dotenv import load_dotenv
# 加载.env文件中的环境变量
load_dotenv()
# 检查API密钥是否已设置
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key or api_key == "your-actual-api-key":
print("错误：请设置有效的DASHSCOPE_API_KEY环境变量")
print("您可以在.env文件中添加：DASHSCOPE_API_KEY=your-actual-api-key")
exit(1)
# 设置API Key
os.environ["DASHSCOPE_API_KEY"] = api_key
# 初始化模型
llm = Tongyi(
model_name="qwen-turbo",
temperature=0.3, # 中低温度，保证解释准确
max_tokens=1000
)
# 创建提示模板
template = """你是一个编程专家。请详细解释以下{language}代码的功能：
代码：
{code}
请从以下几个方面解释：
1. 代码的主要功能
2. 关键步骤的作用
3. 可能的应用场景
解释："""
prompt = PromptTemplate(
input_variables=["language", "code"],
template=template
)
# 使用新的LCEL语法创建链（替代已弃用的LLMChain）
explain_chain = prompt | llm | StrOutputParser()
# 使用示例
code_example = """
def fibonacci(n):
if n <= 1:
return n
return fibonacci(n-1) + fibonacci(n-2)
"""
result = explain_chain.invoke({
"language": "Python",
"code": code_example
})
print(result)这段代码实现了斐波那契数列的计算：
1. **主要功能**：计算斐波那契数列的第n项
- 斐波那契数列：0, 1, 1, 2, 3, 5, 8, 13...
- 每一项等于前两项之和
2. **关键步骤**：
- 基础情况：当n<=1时，直接返回n
- 递归调用：返回fibonacci(n-1) + fibonacci(n-2)
3. **应用场景**：
- 数学计算
- 算法学习
- 递归思想演示
注意：这种实现方式简洁但效率较低，大数值时建议使用动态规划优化。自动提取长文本的核心内容。
from langchain_community.llms import Tongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
# 检查API密钥
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key or api_key == "your-actual-api-key":
print("警告：请设置有效的DASHSCOPE_API_KEY环境变量")
exit(1)
# 初始化模型
llm = Tongyi(
model_name="qwen-turbo",
temperature=0.5, # 中等温度，平衡准确性和可读性
max_tokens=500
)
# 创建提示模板
template = """请为以下文章生成一个简洁的摘要，要求：
1. 提取核心观点
2. 保留关键信息
3. 字数控制在100字以内
文章：
{article}
摘要："""
prompt = PromptTemplate(
input_variables=["article"],
template=template
)
# 创建链 - 使用LCEL语法替代LLMChain
summary_chain = prompt | llm | StrOutputParser()
# 使用示例
article = """
人工智能（AI）技术正在快速发展，深度学习、自然语言处理等技术已经
在各个领域得到广泛应用。大型语言模型如GPT、Claude等的出现，
使得AI能够理解和生成人类语言，极大提升了人机交互的体验。
然而，AI技术的发展也带来了一些挑战，包括数据隐私、算法偏见、
就业影响等问题。我们需要在推动创新的技术的同时，关注这些潜在风险，
确保AI技术的健康发展。
"""
result = summary_chain.invoke({"article": article})
print(result)AI技术快速发展，大型语言模型提升了人机交互体验。
但同时也面临数据隐私、算法偏见等挑战，需要在创新与风险管理间取得平衡。请完成这些练习，巩固 LangChain 入门的关键能力：组件理解、Prompt 设计、输出解析、LCEL 编排、流式输出
⚠️ 提示：下面很多题都是 LangChain / LLM 应用开发的面试高频点（Prompt 可维护性、结构化输出、链路可观测、流式体验）。
1. Tongyi 和 ChatTongyi 有什么区别？你会在什么场景下选它们？
请用 3 句话解释它们的输入输出形态差异，并给出一个“多轮对话”与一个“单次生成”的使用场景。
2. temperature / top_p 的区别是什么？怎么调参更“工程化”？
给出一条规则：什么业务用低温度？什么业务用高温度？你会怎么做 A/B 评估？
3. PromptTemplate 和 ChatPromptTemplate 的差异是什么？
分别写出它们适配的模型输入类型，并说明为什么 ChatPromptTemplate 更适合做“可维护的对话系统”。
4. 为什么要用 OutputParser？不用会有什么工程风险？
举例说明：StrOutputParser/JsonOutputParser 分别解决什么问题；以及“模型输出不合规”时你会怎么兜底。
5. LCEL（prompt | llm | parser）到底解决了什么问题？
请从“可读性、可组合、可测试”三个角度回答，并举例说明如何插入一个中间步骤做日志/校验。
要求：请在一个不包含本项目 static/langchain 目录的路径下运行 Python 示例，避免本地目录名 langchain 遮蔽 pip 安装的官方包。
C1. 【编程题】用 LCEL 写一个“翻译链”（中文 → 英文），只输出翻译结果
要求：使用 PromptTemplate + Tongyi + StrOutputParser，并打印最终字符串。
C2. 【编程题】把自然语言“个人信息”解析为 JSON（结构化输出）
要求：使用 JsonOutputParser，并在提示词中注入 format_instructions；输出为 Python 的 dict。
C3. 【编程题】实现“流式输出”到控制台（打字机效果）
要求：开启 streaming=True，使用 llm.stream()，并在控制台逐段打印。
6. 为什么流式输出能显著提升体验？你会在哪些场景开启/关闭 streaming？
请说明：流式对前端/后端的要求，以及“结构化输出/函数调用”场景为何经常不建议流式。
7. 你会如何处理“输出过长”或“上下文超长”导致的报错？
请列出至少 3 种工程手段：截断策略、摘要、分段生成、限制 max_tokens 等。
1.【概念题】为什么要用 LangChain？不用可以吗？
请从“工程抽象、可维护性、可扩展性、可替换性（换模型）”四点回答。
2.【工程题】如何保证“结构化输出”在生产环境稳定？
请覆盖：提示词合同、解析与校验、失败重试、降级策略、日志与追踪。
掌握了LangChain基础后，下一章我们将学习对话记忆管理， 让AI能够记住对话历史，实现真正的多轮对话！