一、组件简述

• LLM 大语言模型
  • 配置大模型（模型名称、API密钥、参数等）
  • 通用格式兼容 OpenAI 标准
  • 也支持单独导入各厂商的包，例如通义千问（TongYi）
• 解析器
  • 将大模型返回的结果转换成指定的格式。
  • 支持三种解析器类型：JSON、List、对象
• 提示词模板
  • 两种：PromptTemplate、ChatPromptTemplate
• LCEL 链式
  • 链就是将多个步骤连接起来，按照顺序一步步执行。管道式符号 | 创建链；组件间数据类型自动转换、异步调用、支持流式输出、错误处理、重试、缓存、支持自定义。
  • 案例：
    • chain = prompt 提示词 | llm 对话模型 | parser 解析器
      result = chain.invoke({“city”:“北京”, “region”:“朝阳”}) # 填充提示词模板内容，返回AI回答。
  • 原理
    • 当使用 | 符号，实际上调用的是对象的魔术方法 _or_：
    • a | b 时，Python实际执行：
    • result = a._or_(b)

二、组件描述 & 案例

1. langchain 安装

------ langchain 库 ------
pip install langchain

--------  其他库 ----------
# LangGraph
pip install -U langgraph

# dotenv管理密钥，加载环境变量等【Python库】
pip install python-dotenv

# 千问
pip install -U langchain-community dashscope

# DeepSeek
pip install langchain-deepseek

# 外部资源集成
!pip install langchain_community

# 直接转换为 REST API
pip install "langserve[all]"
# 观测平台
pip install -U langsmith

2. LLM 大模型模型

langchain语言模型主要分为两种写法：

1. LLM 通用模型：输入字符串，输出字符串。
2. ChatModel 对话模型【常用】：输入BaseMessage对象，输出是 AIMessage 对象，但使用时通常取其 content 属性获取文本。

2.1. LLM 模型（旧式）

# LLM.invoke
from langchain_community.llms.tongyi import Tongyi
llm = Tongyi()
print(llm.invoke("你好")) 
# 输出：你好呀！✨ 很高兴见到你！

2.2. ChatModel 对话模型【常用】

OpenAI 兼容各家大模型。

from langchain_openai import ChatOpenAI
# import os
# os.environ["OPENAI_API_KEY"] = "your-api-key"  # 替换为实际密钥

# 初始化模型
llm = ChatOpenAI(
    model="qwen3.6-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="sk-xxx",
    max_tokens=500
)

# 调用
response = llm.invoke("你好")
print(response.content)
# 输出：content='你好！很高兴见到你，有什么可以帮你的吗？😊' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 5, 'total_tokens': 19, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 5}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_08f168e49b_prod0820_fp8_kvcache', 'id': 'b091f984-a197-42cb-94d1-80bfc1bb8631', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None} id='run--924ee09c-7aac-4ee1-b148-613099d36b53-0' usage_metadata={'input_tokens': 5, 'output_tokens': 14, 'total_tokens': 19, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}

2.3. 自定义大模型接口（使用私有化部署的 vLLM 大模型）

很完整的封装案例可用于生产（生产环境可以再加些日志、监控…）

该案例langchain版本可能有问题，使用时注意下。
week04\p12-自定义大模型接口

3. 提示词模板

提示词模板是用于生成提示词的模板。例如，{name}，你好！我是你的朋友小爱同学。就是一个提示词模板。

提示词模板中可以使用变量，变量名由字母、数字和下划线组成，且必须以字母开头。

3.1. PromptTemplate 提示词模板

• 第一种用法 推荐用（简便）

from langchain_core.prompts import PromptTemplate

# 定义模板
prompt = PromptTemplate.from_template(
    "你好{name}，\n 你要查询{city}的景点吗？"
)
# 添加参数
result = prompt.format(name="王明", city="北京")

print(result)
# 输出：你好王明，你要查询北京的景点吗？

• 第二种用法

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate(
    input_variables=["topic", "audience", "tone"], # 显示定义变量
    template="""
    请为{audience}写一篇关于{topic}的文章。
    写作风格应该是{tone}的。
    
    文章要求：
    - 内容准确且有用
    - 结构清晰
    - 适合目标受众
    """
)

formatted_prompt = prompt.format(
    topic="人工智能",
    audience="初学者",
    tone="通俗易懂"
)
print(formatted_prompt)

3.2. ChatPromptTemplate 对话模板

# 使用 ChatPromptTemplate 创建对话模板
from langchain_core.prompts import ChatPromptTemplate

# 创建包含 system 和 human 消息的聊天模板
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个名为 {ai_name} 的有用 AI 助手。"),
    ("human", "你好，我的名字是 {name}。{issue}")
])

# 格式化消息
messages = chat_prompt.format_messages(
    ai_name="AI助理",
    name="王明", 
    issue="今天天气怎么样？"
)

for message in messages:
    print(f"{message.type}: {message.content}")
	# 输出：
	# system: 你是一个名为 AI助理 的有用 AI 助手。
	# human: 你好，我的名字是 王明。今天天气怎么样？

3.3. 自定义模板

有复杂提示词参数时使用
week04\p11-自定义模板.ipynb

4. 输出解析器

输出解析器是 LangChain 中的重要组件，用于将大语言模型(LLM)的原始文本输出转换为结构化数据格式，便于程序进一步处理和使用。

主要功能

• 格式化输出: 将 LLM 的自然语言输出解析为 JSON、XML、列表等结构化格式
• 类型转换: 将文本数据转换为 Python 对象（字典、列表、自定义类等）
• 数据验证: 确保输出符合预期的格式和约束条件
• 错误处理: 处理解析失败的情况，提供重试机制

---

以下为常见的输出解析器类型案例：

• StructuredOutputParser = JSON字典解析器
• CommaSeparatedListOutputParser = LIst 列表解析器
• PydanticOutputParser = 对象解析器

解析器的创建流程：定义响应数据结构 -> 创建解析器 -> 获取格式指令 -> 嵌入提示词模板

4.1. JSON 格式解析器 StructuredOutputParser

例如希望大模型返回这样的 JSON 格式数据：
{
“answer”: “巴黎”,
“source”: “地理知识”
}

from langchain.output_parsers import StructuredOutputParser, ResponseSchema
from langchain.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi

# 第1步. 定义响应数据结构【明确指定模型返回的JSON数据结构及参数说明】
response_schemas = [
	# 参数说明：name：JSON中的key（字段名）, description：告诉 LLM 这个字段应该包含什么内容（作为提示的一部分）
    ResponseSchema(name="answer", description="问题的答案"),  
    ResponseSchema(name="source", description="答案来源")
]

# 第2步. 创建解析器（StructuredOutputParser - 结构化输出解析器）
parser = StructuredOutputParser.from_response_schemas(response_schemas)

# 第3步. 获取格式指令：生成一段文本，告诉大模型应该以什么JSON格式返回数据
#    这段文本会被嵌入到提示词中，指导大模型按我们定义的结构输出
format_instructions = parser.get_format_instructions()

# 第4步. 在 prompt 中使用（将解析器的格式嵌入到提示词模板中）
prompt = PromptTemplate(
    template="回答问题：{question}\n\n{format_instructions}", # {format_instructions} 是占位符
    input_variables=["question"],  # 声明：question 是运行时需要传入的变量
    partial_variables={"format_instructions": format_instructions}  # 预填充：将解析器的格式指令绑定到 {format_instructions} 占位符
)

# 初始化大模型【千问大模型，环境变量需要指定 DASHSCOPE_API_KEY=你的API-KEY】
llm = Tongyi(temperature=0)
# 创建链 chain
chain = prompt | llm | parser
# 调用
result = chain.invoke({"question": question})

# 取数据
print(f"问题的答案：{result['answer']}")
print(f"答案来源：{result['source']}")

4.2. List 列表解析器 CommaSeparatedListOutputParser

from langchain.output_parsers import CommaSeparatedListOutputParser
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 第1步. 创建列表解析器（不需要定义结构！）
parser = CommaSeparatedListOutputParser()

# 第2步. 获取格式指令
format_instructions = parser.get_format_instructions()

# 第3步. 创建提示词模板
prompt = PromptTemplate(
    template="列出{subject}的3个要点：\n\n{format_instructions}",
    input_variables=["subject"],
    partial_variables={"format_instructions": format_instructions}
)

# ... 最后大模型返回的就是List列表

3.3. 对象解析器 PydanticOutputParser

from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# 1. 定义结构（Pydantic模型）
class PersonInfo(BaseModel):
    name: str = Field(description="人物姓名")
    age: int = Field(description="年龄")
    occupation: str = Field(description="职业")

# 2. 创建解析器
parser = PydanticOutputParser(pydantic_object=PersonInfo)

# 3. 获取格式指令
format_instructions = parser.get_format_instructions()

# 4. 创建提示词模板
prompt = PromptTemplate(
    template="生成一个虚构人物\n\n{format_instructions}",
    input_variables=[],
    partial_variables={"format_instructions": format_instructions}
)

# 5. 调用大模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
chain = prompt | llm | parser

# 6. 获取结果
result = chain.invoke({})
print(result)
print(f"姓名：{result.name}，年龄：{result.age}，职业：{result.occupation}")

3.4. 自定义输出解析器

将大模型返回的结果封装成指定格式。
week04\p13-自定义输出解释器.ipynb

5. LLM 参数设置 & 添加标签（用于日志追踪）

• LLM 参数设置：同一个模型实例在不同场景使用时，配置不同参数。
• tags 标题：添加日志方便追踪。

# 给 chain 设置模型参数
# 同一个模型实例在不同场景使用时，配置不同参数
chain = prompt | llm | parser
chain = chain.bind(temperature=0.3, max_tokens=512)

-----

# 添加标签用于日志追踪
# LangChain 官方推荐的日志 / 追踪方式
chain = prompt | llm | parser 
chain = chain.with_tags(["RAG服务", "线上环境", "v2.5"]) # 加标签，日志输出
result = chain.invoke({"question": "你好"})

-----

# 添加元数据（尽量不动元数据）
chain = chain.with_metadata({"version": "1.0", "author":"team"})

6. invoke同步、ainvoke异步、batch批量异步调用大模型

注：这里所有的 await（包括 async for、async with）都会挂起当前函数，等执行完成后才会往下执行（不会出现异步还没执行完，但是函数执行结束的情况）。

• invoke() ：同步执行大模型，执行完成后一次性返回完整结果。
• ainvoke()：异步执行大模型，执行完成后一次性返回完整结果。
• batch()：批量异步执行多个问题让一个大模型回答
• stream()：同步流式调用。
• astream()：异步流式调用。

week04\p14-链的解析.ipynb

# 伪代码...
# 构建链
chain = prompt | llm | parser 

# ========== 基础调用 ==========

# 同步调用 - 阻塞当前线程
result = chain.invoke({"city":"北京", "region":"朝阳"})

# 异步调用 - 挂起当前协程，不阻塞事件循环
# 注：await 会挂起当前函数，等待执行完成后才会往下执行，但不会阻塞事件循环
result = await chain.ainvoke({"city":"北京", "region":"朝阳"})


# ========== 批量调用 ==========

# 同步批量 - 阻塞当前线程（内部用线程池实现并发）
# 注：会阻塞主线程，但任务之间是并发的
results = chain.batch([input1, input2, input3])

# 异步批量 - 不阻塞事件循环，并发执行
# 注：挂起当前函数，等待所有任务完成才会往下执行，但等待期间可执行其他协程
results = await chain.abatch([input1, input2, input3])


# ========== 流式调用 ==========

# 同步流式 - 阻塞当前线程，边生成边输出
for chunk in chain.stream({"city": "北京", "region": "朝阳"}):
    print(chunk, end="", flush=True)

# 异步流式 - 不阻塞事件循环，边生成边输出
# 注：挂起当前函数，等待所有任务完成才会往下执行，但等待期间可执行其他协程
async for chunk in chain.astream({"city": "上海", "region": "浦东"}):
    print(chunk, end="", flush=True)

7. 上下文记忆（短记忆、Mem0长记忆）

LangChain 上下文记忆系统 - 短时记忆、Mem0长期记忆增强框架

8. 工具调用（普通模式、ReAct思维链+行动）

8.1. Agent 使用工具调用

工具调用案例（普通调用案例、ReAct模式调用案例）

8.2. LangChain 内置工具

LangChain内置了很多好用的工具（搜索引擎、数据库操作、代码解释器等）【点击跳转查看内置工具文档】

以下是使用搜索工具 DuckDuckgoSearch 的案例：

# 安装搜索工具 duckduckgo-search
!pip install -qU duckduckgo-search langchain-community ddgs

# 基本搜索方式，返回 String 文本
from langchain_community.tools import DuckDuckGoSearchRun

search = DuckDuckGoSearchRun() # 基本模式
search.invoke("苹果公司的创始人 ?") # 执行搜索

# 输出：'2025年7月22日 — 1976年4月，史蒂夫·賈伯斯、史蒂夫·沃茲....

# 获取更多附加信息（例如链接、来源）,返回 JSON 格式
from langchain_community.tools import DuckDuckGoSearchResults

search = DuckDuckGoSearchResults(output_format="list") # 溯源模式
search.invoke("苹果公司的创始人 ?") # 执行搜索

# 输出
[{'snippet': '2025年7月22日 — 1976年4月，史蒂夫·賈伯斯、史蒂夫·沃茲尼亞克和羅納德·韋恩創立了蘋果公司，目的是為了研發和銷售沃茲尼亞克Apple I個人電腦，但韋恩12天後就放棄了自己的股份。',
  'title': '蘋果公司- 維基百科，自由的百科全書',
  'link': 'https://zh.wikipedia.org/zh-hk/蘋果公司'},
 {'snippet': '2025年7月13日 — 1976年，沃茲尼克與史蒂夫·賈伯斯合夥創立蘋果電腦公司，並在1970年代末研發出第一代和第二代蘋果電腦。 二代電腦一經發售即風靡全美，成為1970-80年代之交銷量最佳的個人電 ...',
  'title': '史蒂夫·沃茲尼克',
  'link': 'https://zh.wikipedia.org/zh-hk/史蒂夫·沃兹尼亚克'}]

9. 普通方法附加重试能力（Runnable 套壳）

# 普通方法
def query_order(order_id: str) -> dict:
    # 查数据库、调 API 等
    return {"order_id": order_id, "status": "ok"}
   
runner = RunnableLambda(query_order).with_retry() # 套壳 + 抛异常时重试，默认共 3 次
result = runner.invoke("20251114001") # 传入参数，并执行方法

三、链的常用处理（类型定义[AOP增强]、并行跑多个大模型、错误处理、动态配置大模型参数）

1. 类型定义 - 类似AOP上下文增强

常用用途：清洗用户输入数据、封装大模型返回结果。

# 明确类型定义（输入什么类型、输出什么类型）
## 输入：一个字典 {"key": "字符串"}
## 输出：一个字典 {"key": 任意类型}
class MyChain(Runnable[Dict[str, str], Dict[str, Any]]):
    pass

# 完整案例
from typing import Dict, Any, Optional
from langchain_core.runnables import Runnable, RunnableConfig

# 1. 定义类型安全的链（接收 dict 输入，输出 dict）
class MyChain(Runnable[Dict[str, str], Dict[str, Any]]):
    """一个简单的自定义链：提取输入中的 name 字段，返回问候语"""
   	# 核心参数 
   	# - input 用户输入。
   	# - config 接收回调、标签、元数据等运行时配置
    def invoke(self, input: Dict[str, str], config: Optional[RunnableConfig] = None, **kwargs: Any) -> Dict[str, Any]:
        # 从输入字典中获取 name，如果没有则使用默认值
        name = input.get("name", "访客")
        
        # 处理业务逻辑
        greeting = f"你好,{name}! 欢迎使用langchain."
        
        # 返回字典格式的输出
        return {
            "greeting": greeting,  # 你好,王明! 欢迎使用langchain.
            "input_name": name,    # 王明
            "status": "success"
        }

# 2. 使用方式一：直接调用
chain = MyChain()
result = chain.invoke({"name": "王明"})
print(result)
# 输出: {'greeting': '你好,王明! 欢迎使用langchain.', 'input_name': '王明', 'status': 'success'}

# 3. 使用方式二：通过 LCEL 管道与其他组件串联
from langchain_core.runnables import RunnableLambda

# 先做预处理：提取用户输入
preprocess = RunnableLambda(lambda x: {"name": x["user_input"].upper()})
# 管道组合：预处理 -> 自定义链
pipeline = preprocess | chain
# 执行整个管道
final_result = pipeline.invoke({"user_input": "bob"})
print(final_result)
# 输出: {'greeting': 'Hello, BOB! Welcome to LangChain.', 'input_name': 'BOB', 'status': 'success'}

# 4. 批量处理（自动支持）
batch_results = chain.batch([
    {"name": "Alice"},
    {"name": "Bob"},
    {"name": "Charlie"}
])
print(batch_results)
# 输出: 三个结果的列表

2. 并行跑多个大模型【一个问题让多个大模型回答】

调用 .invoke() 后会等所有大模型都跑完后主线程才会往下走。

# 并行（同步跑多个大模型）
parallel_processing = RunnableParallel({
    "fast_analysis": quick_model,
    "detailed_analysis": detailed_model
})

# 完整案例
from langchain_core.runnables import RunnableParallel
from langchain_ollama import ChatOllama

# 两个模型
fast_model = ChatOllama(model="qwen2.5:7b")
detailed_model = ChatOllama(model="qwen2.5:14b")

# 并行执行【重点】
parallel = RunnableParallel({
    "快模型回答": fast_model,
    "慢模型回答": detailed_model
})

# 调用（两个模型同时跑）
result = parallel.invoke("什么是RAG？") # 等待所有并行任务都完成后，才继续往下执行。

# 取不同模型结果【重点】
print(result["快模型回答"])
print(result["慢模型回答"])

3. 错误处理（重试 | 切换备选大模型）

应用场景：当主大模型运行异常或触发限流时，系统将自动切换至备用大模型
案例1：主模型报错时自动切换至备用大模型
案例2：主模型重试3次失败后自动切换至备用大模型

案例 1

# 添加错误处理（大模型运行错误后切换到备用大模型）
robust_chain = (
    preprocessing |
    model.with_fallbacks([backup_model]) | # 当大模型出现异常时，切换到备用大模型 backup_model
    postprocessing
)

# 完整案例
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_models import ChatTongyi

# 设置通义千问的 API key（请替换为你的真实 key）
import os
os.environ["DASHSCOPE_API_KEY"] = "your-dashscope-api-key-here"

# 1. 主模型：qwen-turbo
main_model = ChatTongyi(
    model="qwen-turbo",
    temperature=0.7,
)

# 2. 备用模型：qwen-plus（更强大，作为 fallback）
backup_model = ChatTongyi(
    model="qwen-plus",
    temperature=0.5,
)

# 3. 后处理：解析输出
postprocessing = StrOutputParser()

# 4. 构建带 fallback 的链（使用 f-string 风格的简单提示词）
robust_chain = (
    ChatPromptTemplate.from_template("{input}")  # 最简单的提示词模板
    | main_model.with_fallbacks([backup_model])
    | postprocessing
)

# 5. 运行测试
if __name__ == "__main__":
    # 正常调用
    result = robust_chain.invoke({"input": "你好，请介绍一下你自己"})
    print("回答:", result)
    
    # 你也可以测试 fallback 机制（通过模拟主模型失败）
    # 注意：正常情况下两个模型都可用，不会触发 fallback

案例 2

# 添加错误处理（主大模型错误重试 + 切换备用大模型）
robust_llm = llm.with_retry(
    stop_after_attempt=3,
    retry_if_exception_type=(RateLimitError,APIConnectionError)
).with_fallbacks([fallback_llm, local_llm]) # 两个备选大模型

# 完整案例
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_models import ChatTongyi
from openai import RateLimitError, APIConnectionError  # 通义千问也使用相同的异常类型

# 设置通义千问的 API key
import os
os.environ["DASHSCOPE_API_KEY"] = "your-dashscope-api-key-here"

# 1. 主模型：qwen-turbo（重试3次）
main_model = ChatTongyi(
    model="qwen-turbo",
    temperature=0.7,
    max_retries=0  # 关闭内置重试
).with_retry(
    stop_after_attempt=3,  # 重试3次
    retry_if_exception_type=(RateLimitError, APIConnectionError)  # 只对限流/网络错误重试
)

# 2. 备用模型：qwen-plus（作为 fallback）
backup_model = ChatTongyi(
    model="qwen-plus",
    temperature=0.5,
)

# 3. 后处理：解析输出
postprocessing = StrOutputParser()

# 4. 构建链：重试3次 + 备用模型
robust_chain = (
    ChatPromptTemplate.from_template("{input}")
    | main_model.with_fallbacks([backup_model])  # 主模型重试3次失败后，切到备用（备用可以设多个）
    | postprocessing
)

# 5. 运行测试
if __name__ == "__main__":
    # 正常调用
    result = robust_chain.invoke({"input": "你好，请介绍一下你自己"})
    print("回答:", result)

4. 大模型动态参数配置（配置管理）

动态修改大模型参数。
场景：A/B 测试、多租户应用（不同用户不同配置）、动态调整创作风格等。

# 配置管理（动态配置大模型参数）
configurable_chain = model.configurable_fields(
    temperature=ConfigurableField(id="my_temperature", name="Temperature")
)

# 完整案例
from langchain_openai import ChatOpenAI
from langchain_core.runnables.config import ConfigurableField
from langchain_core.messages import HumanMessage

# 定义模型（设置默认值）
model = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.5,
    max_tokens=50
)

# 1.同时暴露两个参数可以动态调整
configurable_chain = model.configurable_fields(
	# 返回值 temperature 是指"模型的 temperature 参数"
    temperature=ConfigurableField(
        id="my_temperature", # ← 动态修改参数时用这个
        name="Temperature"   # ← 命名
    ),
    # 返回值 max_tokens 是指"模型的 max_tokens 参数"
    max_tokens=ConfigurableField(
        id="my_max_tokens",  # ← 动态修改参数时用这个
        name="Max Tokens"    # ← 命名
    )
)

# 2. 定义消息列表（用户问题）
messages = [HumanMessage(content="讲个笑话")]

# 3. 使用默认配置调用大模型
response_default = configurable_chain.invoke(messages)
print(f"默认 (temp=0.5, max=50): {response_default.content}\n")

# 4. 调用大模型，同时修改参数
# ⚠️ 注意：with_config 中对应 ConfigurableField 时设置的暴漏参数的 id
response_custom = (configurable_chain.with_config(configurable={
        "my_temperature": 0.9,
        "my_max_tokens": 100
    })
    .invoke(messages)  # ← 消息列表
)
print(f"自定义 (temp=0.9, max=100): {response_custom.content}")

5. batch 批量调用大模型

四、将Langchain封装成网络服务 - 接口

week04\p9-封装为网络服务.ipynb

项目文件结构：
chat_service/
├── main.py # FastAPI 主应用
├── models.py # 数据模型定义
├── chat_chain.py # 对话链逻辑
├── session_manager.py # 会话管理
└── requirements.txt # 依赖包