《LangChain 工程化入门：如何像搭乐高一样构建 AI 应用？》

前言

笔记基于easy-langent/DataWhale进行总结，详细部分可以进入阅读学习，当前笔记是针对大模型应用

欢迎给我的Agent学习仓库提出修改建议：AI-Agent-learning

最近因为比赛需要，尝试使用智谱ai低代码平台，里面的拖拽式编程就有点头疼：

  1. 不会写prompt：LLM解决问题的能力一般，写了很长篇幅，但有时出现直接把System的提示词直接输出的情况
  2. AI整个处理流程不清晰：在一个业务中参杂了一些不是核心的组件

从中意识到提示词的重要性，而LangChain的工程化思维是当前较大的收获

LangChain解决方案

大模型开发痛点

接口不统一

提示词难管理

输出不可控

统一模型抽象 ChatModel/LLM

提示词模板 PromptTemplate

输出解析器 OutputParser

屏蔽底层差异

参数化 & 复用

结构化数据

核心组件一：统一模型调用

市面上的海量大模型调用的接口都不同：Qwen、Deepseek，那需要根据使用模型去不断修改代码吗？

LangChain中的“模型抽象”组件就是为了解决不同厂商的模型封装了统一的调用接口

屏蔽底层细节，专注应用逻辑

（1）LLM vs ChatModel

LLM（文生文）:Hugging Face的LIama、Qwen，适合简单的文本生成

ChatModel：GPT-4o、DeepSeek-Chat接收一系列“对话消息”，返回一条对话消息

（2）实操：统一接口调用不同模型

DataWhale中用的是DeepSeek做演示，本篇及后面都会使用Qwen-turbo大模型进行大模型应用，会有部分代码不同

  • 调用OpenAI的ChatModel

    from langchain_openai import ChatOpenAI
#⬆️OpenAI统一对话接口 
from dotenv import load_dotenv
import os
load_dotenv()#👈加载API密钥（从.env文件读取）

API_KEY=os.getenv('OPENAI_API_KEY')
BASE_URL=os.getenv('BASE_URL')

if not API_KEY:
    raise ValueError("未检测到API_KEY，请检查.env文件是否配置正确")

# 1. 初始化对话模型
# 不管是哪个厂商的ChatModel，初始化参数都类似（model、temperature等）
chat_model = ChatOpenAI(
    api_key=API_KEY,
    base_url=BASE_URL,
    model="qwen-turbo", #👈这里我选的是Qwen-turbo
    temperature=0.5,
    max_tokens=50
)

# 2. 构造对话消息
# ChatModel需要接收的是“消息列表”，每个消息有角色（user/assistant/system）和内容
message =[
    {"role": "system","content":"你是一个耐心的AI学习助手，回复简洁易懂，适合高校学生理解。"},
    {"role": "user","content":"请用3句话解释什么是LangChain"}
]

# 3. 调用模型生成结果
# 统一调用方法：invoke()，传入消息列表
result = chat_model.invoke(message)

print("ChatModel回复：")
print(result.content)

    结果

    在这里插入图片描述

    角色的分配本质是“约束层级”，不是“说话人”

    我看成：“system 是黑板上的教学要求，user 是学生提问，assistant 是老师刚讲过的内容——三者构成上下文连续性。”

    当然：

    system、user、assistant 这三个角色的设计，远不止是简单的消息分类。它实际上构建了一个对话的上下文沙盒。system 是沙盒的边界和规则，user 是驱动沙盒内状态变化的输入，而 assistant 则是沙盒当前状态的快照。理解了这一点，我们就能更精准地控制模型的行为，而不是把它当作一个黑盒随意投喂问题。

    设定全局规则

    提供新任务

    构成历史上下文

    System
    (系统指令)

    User
    (用户输入)

    Assistant
    (助手回复)

核心组件二：提示词模板**（PromptTemplate）**

我看成：打比方，我写一道数学答题，你可以用不同的方法去解决，但是最后的相同的是输出正确答案（只不过在大模型学习中，AI生成的答案每次都不太相同）

（PromptTemplate）：把固定的提示文本和动态参数分离，这样可以进行复用

（1）基础模版：标准化提示与动态参数

就是定义一个带“动态参数的模板”，减少出错，便于复用和维护

# 导入PromptTemplate
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

load_dotenv()
API_KEY = os.getenv("OPENAI_API_KEY")
BASE_URL = os.getenv("BASE_URL")  # 从环境变量读取，未配置时默认为None（使用OpenAI官方地址）

if not API_KEY:
    raise ValueError("未检测到 API_KEY，请检查 .env 文件是否配置正确")

chat_model = ChatOpenAI(
    api_key=API_KEY,
    base_url=BASE_URL,
    model="qwen-turbo",  # 选择对话模型
    temperature=0.5,        # 随机性：0-1，越小越严谨，越大越有创造力
    max_tokens=100          # 最大生成 tokens 数，避免生成过长内容
)
# 1. 定义提示词模板
# input_variables：动态参数列表（这里是user_role和subject）
# template：提示词模板字符串，用{参数名}表示动态参数
prompt_template = PromptTemplate(
    input_variables=["user_role", "subject"],
    template="请给{user_role}写一段50字左右的{subject}学习建议，语言简洁实用，分2个小要点。"
)

# 2. 格式化模板（传入具体参数，生成完整提示词）
# 给“高校学生”生成“LangChain”学习建议
formatted_prompt = prompt_template.format(
    user_role="高校学生",
    subject="LangChain"
)
print("格式化后的提示词：")
print(formatted_prompt)

# 3. 调用模型生成结果
result = chat_model.invoke([{"role": "user", "content": formatted_prompt}])

print("\n生成的学习建议：")
print(result.content)

结果
在这里插入图片描述

（2）提示词模板进阶用法：少样本提示模板

有时简单的提示词模板不能够让模型生成令人满意的答案，这时候就需要有人打个样–“少样本提示”，LangChain的FewShotPromptTemplate就是专门实现这个功能的

# 导入必要的模板类
from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

load_dotenv()
API_KEY = os.getenv("OPENAI_API_KEY")
BASE_URL = os.getenv("BASE_URL")  # 从环境变量读取，未配置时默认为None（使用OpenAI官方地址）

if not API_KEY:
    raise ValueError("未检测到 API_KEY，请检查 .env 文件是否配置正确")

chat_model = ChatOpenAI(
    api_key=API_KEY,
    base_url=BASE_URL,
    model="qwen-turbo",  # 选择对话模型
    temperature=0.5,        # 随机性：0-1，越小越严谨，越大越有创造力
    max_tokens=200          # 最大生成 tokens 数，避免生成过长内容
)

with open("learning_method_examples.json","r",encoding="utf-8") as f:

# 1. 定义示例（少样本的核心：给模型看的参考案例）
    examples = [
      	{
          "subject": "Python编程",
          "method": "核心目标：掌握基础语法和常用库；学习步骤：1. 学习变量、函数等基础语法 2. 实操小项目（如计算器） 3. 学习Pandas、Matplotlib库；注意事项：多动手实操，遇到错误及时调试。"
      	},
      	{
        "subject": "机器学习",
        "method": "核心目标：理解基础算法原理和应用场景；学习步骤：1. 复习数学基础（线性代数、概率） 2. 学习经典算法（线性回归、决策树） 3. 用Scikit-learn实操；注意事项：先理解原理，再动手实现，避免死记硬背。"
    	}
	]

# 2. 定义示例模板（告诉模型如何解析示例）
example_template = """
学科：{subject}
学习方法：{method}
"""
example_prompt = PromptTemplate(
    input_variables=["subject", "method"],
    template=example_template
)

# 3. 定义最终的提示词模板（包含示例和用户需求）
few_shot_prompt = FewShotPromptTemplate(
    examples=examples,                # 传入示例
    example_prompt=example_prompt,    # 示例模板
    suffix="学科：{new_subject}\n学习方法：",  # 最终给用户的提示（在示例之后）
    input_variables=["new_subject"]   # 动态参数：用户要查询的新学科
)

# 4. 格式化模板（传入新学科：LangChain）
formatted_prompt = few_shot_prompt.format(new_subject="LangChain")
print("少样本提示词：")
print(formatted_prompt)

# 5. 调用模型生成结果
result = chat_model.invoke([{"role": "user", "content": formatted_prompt}])

print("\n生成的LangChain学习方法：")
print(result.content)

（3）少样本提示模板的高效管理

在工程中会有比上面代码中example更多的示例、需要动态更新、按条件筛选示例等问题，这部分来解决FewShotPromptTemplate的痛点

痛点
  • 示例过多导致提示词冗长：使用ExampleSelector动态筛选相关示例，避免冗余
  • 示例维护困难：将示例存储在文件（JSON/CSV）中，批量加载与更新
  • 不同场景需要不同示例：按输入参数匹配示例类型，实现场景化示例分发

这里通过使用LangChain提供的ExampleSelector组件进行解决

ExampleSelector可以根据输入的动态参数筛选最相关的示例，减少提示词体积，提升大模型的响应效率

# 要先在powershell运行 pip install ChatTongyi
# 这里我是用Qwen的大模型
from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate
from langchain_core.example_selectors import BaseExampleSelector
from langchain_community.chat_models import ChatTongyi  # ← 关键修改
from dotenv import load_dotenv
import os
import json
from typing import Dict, List

# 环境初始化
load_dotenv()
DASHSCOPE_API_KEY = os.getenv("OPENAI_API_KEY")
if not DASHSCOPE_API_KEY:
    raise ValueError("未检测到 DASHSCOPE_API_KEY，请在 .env 中配置")

# 使用 Tongyi 模型（显式传递 DASHSCOPE_API_KEY）
chat_model = ChatTongyi(
    dashscope_api_key=DASHSCOPE_API_KEY,  # 显式传递 API Key
    model="qwen-turbo",
    temperature=0.4,
    max_tokens=200
)
# 2. 工程化示例管理：从JSON文件加载示例（避免硬编码，便于维护）
with open("learning_method_example.json", "r", encoding="utf-8") as f:
    examples = json.load(f)  # 从JSON中直接提取示例数据列表
# 示例文件格式参考（learning_method_examples.json）前面内容

# 3. 方案A：ExampleSelector按长度筛选示例（控制提示词总长度）
# example_selector = LengthBasedExampleSelector(
#     examples=examples,
#     example_prompt=PromptTemplate(
#         input_variables=["subject", "difficulty", "method"],
#         template="学科：{subject}\n难度：{difficulty}\n学习方法：{method}\n"
#     ),
#     max_length=150,  # 控制示例总长度，避免提示词过长
#     get_text_length=lambda x: len(x)  # 长度计算函数
# )

# 4. 方案B（推荐）：自定义ExampleSelector按难度筛选示例
# 当需要根据用户输入的特征（如难度）精准匹配示例时，可自定义ExampleSelector
class DifficultyExampleSelector(BaseExampleSelector):
    """根据用户输入的 difficulty 字段筛选样本"""
    def __init__(self, examples: List[Dict[str, str]]):
        self.examples = examples

    def add_example(self, example: Dict[str, str]) -> None:
        self.examples.append(example)

    def select_examples(self, input_variables: Dict[str, str]) -> List[Dict]:
        # 获取用户输入的难度等级，如果没有提供则默认为 'easy'
        target_difficulty = input_variables.get("difficulty", "easy")
        # 过滤出匹配难度的所有示例
        return [ex for ex in self.examples if ex.get("difficulty") == target_difficulty]


# 本案例使用方案B（按难度筛选），如需使用方案A（按长度筛选），取消方案A的注释并注释掉方案B即可
example_selector = DifficultyExampleSelector(examples=examples)

# 5. 构建工程化少样本模板
few_shot_prompt = FewShotPromptTemplate(
    example_selector=example_selector,  # 替换固定examples为动态选择器
    example_prompt=PromptTemplate(
        input_variables=["subject", "difficulty", "method"],
        template="学科：{subject}\n难度：{difficulty}\n学习方法：{method}\n"
    ),
    example_separator="\n", # 控制examples示例之间的分隔方式
    prefix="少样本提示：",
    suffix="参考以上示例，回答：\n学科：{new_subject}\n难度：{difficulty}\n学习方法：",
    input_variables=["new_subject", "difficulty"]  # 新增难度参数
)

# 6. 动态生成不同难度的提示词
# 场景1：生成入门级LangChain学习方法
formatted_prompt_easy = few_shot_prompt.format(
    new_subject="LangChain",
    difficulty="easy"
)
print("入门级少样本提示词：")
print(formatted_prompt_easy)
result_easy = chat_model.invoke([{"role": "user", "content": formatted_prompt_easy}])
print("\n入门级学习方法：")
print(result_easy.content)

# 场景2：生成进阶级LangChain学习方法
formatted_prompt_hard = few_shot_prompt.format(
    new_subject="LangChain",
    difficulty="hard"
)
print("\n进阶级少样本提示词：")
print(formatted_prompt_hard)
result_hard = chat_model.invoke([{"role": "user", "content": formatted_prompt_hard}])
print("\n进阶级学习方法：")
print(result_hard.content)

1）示例从JSON文件加载，便于批量维护；

2）通过自定义ExampleSelector按难度动态筛选示例，适配不同学习阶段需求（也可切换为LengthBasedExampleSelector按长度控制）；

3）控制示例总长度，避免触发模型token限制。

我的思考：将提示词（Prompt）视为代码的一部分，是迈向工程化的关键一步。PromptTemplate 让我们能够像管理函数参数一样管理提示词的动态部分，而 FewShotPromptTemplate 则像是为模型编写“单元测试用例”。这种做法极大地提升了提示词的可维护性和可测试性，让提示工程从一门玄学，变成了一项可以被版本控制、Code Review 和持续集成的工程实践。

输出解析：让输出更可控

PromptTemplate 和 FewShotPromptTemplate可以有效解决 “如何向模型提供结构化、规范化的格式指令” 的问题；但输出解析（Output Parsing / Output Control）是要解决的是另一个问题：

如何将大模型返回的非结构化自然语言转化为程序可直接处理的结构化数据。

WHY？为什么需要输出解析层？

在工程实践中，大模型天然倾向于输出自由流畅的自然语言文本。这虽然对人类阅读很友好，但对程序就很麻烦。想象一下这两种场景：

  1. 无解析器（原始输出）：

    “这个手机续航好、拍照清晰，但系统有点卡”

    如果你想提取关键词做分析，就必须写复杂的正则表达式或字符串处理逻辑，而且每次模型输出格式稍有变化（比如加个表情符号 😊），你的代码就可能崩溃。

  2. 有解析器（结构化输出）：

    ["续航好", "拍照清晰", "系统卡顿"]

    这是一个标准的 Python 列表，你可以直接用 for 循环遍历、存入数据库或进行后续计算，稳定、可靠、高效。

输出解析层的核心价值就在于此：它充当了“大模型”与“业务系统”之间的翻译官和质检员。

场景需求 无解析器（原始输出） 有解析器（结构化输出）
提取商品评论关键词 “这个手机续航好、拍照清晰…” ["续航好", "拍照清晰", ...]
生成用户信息 “用户叫张三，25岁…” {"name": "张三", "age": 25}
分析订单状态 “订单号A123已发货…” [{"orderId": "A123", "status": "已发货"}, ...]

HOW？LangChain 如何实现输出解析？

LangChain 提供了多种开箱即用的 OutputParser，我们可以根据项目阶段和需求灵活选择。

(1) StrOutputParser：最基础的“净化器”

适用场景：当你只需要纯文本，并希望将模型返回的 AIMessage 对象统一转换为标准的 Python 字符串 (str) 时。

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 初始化模型
llm = ChatOpenAI(model="qwen-turbo", api_key=API_KEY, base_url=BASE_URL)

# 创建解析器
parser = StrOutputParser()

# 链式调用
chain = llm | parser
result = chain.invoke("请简要介绍 LangChain")

print(type(result)) # <class 'str'>
print(result)       # 纯文本内容

它的核心作用不是清洗内容，而是确保数据类型的一致性，让后续的字符串操作（如拼接、切片）变得简单直接。

(2) JsonOutputParser：快速原型的“加速器”

适用场景：在开发 Demo 或 MVP（最小可行产品）时，需要快速获得一个简单的键值对结构（如字典），无需复杂的类型校验。

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate

# 创建 JSON 解析器
parser = JsonOutputParser()

# 构建提示模板，自动嵌入格式指令
prompt = PromptTemplate(
    template="请介绍1个LangChain开发工具，输出工具名和核心功能。{format_instructions}",![请添加图片描述](https://i-blog.csdnimg.cn/direct/2dda35df202d40d6a69ee517c3148315.png)

    input_variables=[],
    partial_variables={"format_instructions": parser.get_format_instructions()}
)

chain = prompt | llm | parser
result = chain.invoke({})

print(result) 
# {'tool_name': 'LangSmith', 'core_function': '...'}
print(result['tool_name']) # 可直接访问字段

⚠️ 注意：JsonOutputParser 不会严格校验字段的类型和枚举值，适合快速验证想法，但不适合直接用于生产环境。

(3) PydanticOutputParser：工程上线的“守门员”

适用场景：这是工程化项目的首选方案。通过 Pydantic 定义严格的数据模型，解析器会自动校验模型的输出。任何不符合预定义格式的输出都会被拒绝并抛出异常，从而在早期就规避数据错误的风险。

from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser

# 1. 定义数据模型（这就是你的“合同”）
class ToolInfo(BaseModel):
    tool_name: str = Field(description="工具名称")
    difficulty: str = Field(description="学习难度", enum=["简单", "中等", "复杂"])

# 2. 创建强校验解析器
parser = PydanticOutputParser(pydantic_object=ToolInfo)

prompt = PromptTemplate(
    template="{user_input}。\n{format_instructions}",
    input_variables=["user_input"],
    partial_variables={"format_instructions": parser.get_format_instructions()}
)

chain = prompt | llm | parser
result = chain.invoke({"user_input": "请介绍 LangSmith"})

print(result.tool_name) # 类型安全，IDE有智能提示
print(result.model_dump()) # 转为标准字典
40% 25% 25% 10% OutputParser 选型指南 StrOutputParser(基础文本) JsonOutputParser(快速原型) PydanticOutputParser(生产环境) Custom Parser(特殊需求)

BaseOutputParser —— 一切解析器的“源头”

前面我们介绍了 StrOutputParser、JsonOutputParser 和 PydanticOutputParser 这些开箱即用的工具，而这些工具本质上是 BaseOutputParser —— LangChain 中所有输出解析器的抽象基类（Abstract Base Class）。

你可以把它理解为抽象类。它本身不能直接使用（因为它是抽象的），但它定义了所有具体解析器必须遵守的规则。这个规则主要体现在两个核心方法上：

  1. parse(text: str) -> Any
    • 作用：这是解析器的“心脏”。它接收模型返回的原始字符串 text，然后施展魔法，将其转换成你想要的任何格式（字符串、字典、列表，甚至是自定义对象）。
    • 关键：如果转换失败（比如模型没按要求输出 JSON），你应该在这里抛出一个异常。这个异常可以被 LangChain 的重试机制捕获，从而让模型重新生成一次，极大地提高了应用的鲁棒性。
  2. get_format_instructions() -> str
    • 作用：这是解析器的“说明书”。它会生成一段清晰的文本指令，告诉大模型：“嘿，请你按照这个格式来回答我！”。
    • 妙用：你可以把这个指令通过 partial_variables 轻松地嵌入到你的 PromptTemplate 中，确保模型从源头就朝着正确的方向输出。
什么时候需要自定义？

当你发现内置的解析器都无法满足你的特殊需求时，就是时候祭出 BaseOutputParser 了。

举个例子：你的业务系统要求模型必须用 工具名@核心功能@学习难度 这种特殊的分隔符格式来回答。这时候，你就可以创建自己的解析器：

from langchain_core.output_parsers import BaseOutputParser

class CustomToolParser(BaseOutputParser):
    def parse(self, text: str) -> dict:
        """将模型输出按 '工具名@核心功能@学习难度' 解析为字典"""
        parts = text.strip().split("@")
        if len(parts) != 3:
            # 如果格式不对，果断抛出异常！
            raise ValueError(f"输出格式错误！需满足「工具名@核心功能@学习难度」")
        return {
            "tool_name": parts[0],
            "function": parts[1],
            "difficulty": parts[2]
        }

    def get_format_instructions(self) -> str:
        """生成提示词，引导模型按自定义格式输出"""
        return "请严格按照「工具名@核心功能@学习难度」格式输出，例如：LangSmith@全链路调试监控@中等"

我的思考：BaseOutputParser 的存在，完美体现了 LangChain “约定优于配置” 和 “可扩展性” 的设计哲学。它没有试图穷尽所有可能的解析场景，而是提供了一个简单而强大的接口，让你可以轻松地将自己的业务逻辑无缝集成到 LangChain 的工作流中。这正是一个优秀框架应该具备的特质——既是脚手架，也是乐高底板。

用户输入

PromptTemplate
(构造提示)

ChatModel
(调用模型)

OutputParser
(解析结果)

结构化输出

总结与展望：像搭乐高一样构建 AI 应用

回顾本章，我们围绕大模型应用开发中的三大痛点——接口不统一、提示词难管理、输出不可控——逐一击破。

  • ChatModel 让我们屏蔽了底层模型的差异，无论 Qwen、DeepSeek 还是 OpenAI，调用方式如出一辙。
  • PromptTemplate 和 FewShotPromptTemplate 将提示工程从“艺术”变成了“工程”，通过参数化和示例引导，实现了提示词的复用与规范化。
  • OutputParser 家族（从 StrOutputParser 到 PydanticOutputParser，再到 BaseOutputParser）则为我们架起了一座桥梁，将大模型天马行空的自然语言，稳稳地转化为程序世界里严谨可靠的结构化数据。

这一切的核心思想，就是“组件化”。LangChain 就像一套精心设计的乐高积木，每个组件（Component）都有明确的职责和标准的接口。我们开发者要做的，不是从零开始造轮子，而是学会如何选择合适的积木，并将它们优雅地拼接在一起。

最终目标：让我们的精力从繁琐的底层适配和数据清洗中解放出来，专注于更高价值的应用逻辑和产品创新。

下一章将学习 LangChain 中的Memory（记忆） 和 Tool（工具）。届时，我们的 AI 应用将不再是一个“金鱼记忆”的问答机器，而是一个能记住上下文、能调用计算器、能查询数据库的真正智能体（Agent）！

# 人工智能
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

海外动态代理行业彻底换代，AI 数据时代新标准已到来

avatar AtomGit开源社区
cover

2026年SEO+GEO优化指南：搜索排名机制解析与实用工具推荐

avatar AtomGit开源社区
cover

2026年论文写作AI哪个好用？7款工具实测对比，轻松拿捏

avatar AtomGit开源社区