随着大模型应用普及，越来越多开发者不再满足简单调用 LLM 接口。直接拼接字符串写 Prompt、手动维护对话上下文、自己实现文档检索，代码混乱难以复用，一旦业务复杂，维护成本急剧上升。 LangChain 作为目前最流行的 LLM 应用开发框架，提供标准化组件：模型封装、提示词管理、对话记忆、检索链路、智能体编排。我们不需要从零造轮子，像搭积木一样快速搭建知识库问答、对话机器人、自动化 Agent。因此开启这套 LangChain 从入门到实战系列，全程基于 v1.2 版本。

1. 环境搭建

本系列使用conda作为包管理工具，如果只是通过python进行agent应用开发更推荐使用uv进行包管理

1.1 conda基本命令

shell
conda -help # 查看帮助
conda info # 查看 conda 信息
conda --version # 查看 conda 版本
conda update conda # 更新 Conda (慎用)
conda clean -all # 清理不再需要的包
conda <指令> --help # 查看某一个指令的详细帮助
conda config --show # 查看 conda 的环境配置
conda clean -p # 清理没有用, 没有安装的包
conda clean -t # 清理 tarball
conda clean --all # 清理所有包和 conda 的缓存文件

1.2. 创建conda环境

shell
# 样例 创建一个名为 langchain1.2 的环境, python 版本为3.13.12
conda create --name langchain1.2 python=3.13.12

1.3. 切换 Conda环境

Conda可以创建多套相互隔离的 Python环境，使用 conda activate env_name 可以切换不同的环境。

shell
#语法
conda activate env_name
#样例 切换到 langchain1.2 环境
conda activate langchain1.2

如果要退出此环境，回到基础环境，可以使用如下命令

shell
#退出当前环境
conda deactivate

1.4. Pycharm 新建项目

2. model 的创建

2.1 调用模型提供商API

2.1.1 调用阿里云百炼API

python
from langchain_openai import ChatOpenAI
from langchain_community.chat_models import ChatTongyi
import dotenv
import os


dotenv.load_dotenv()
# llm_qianwen = ChatOpenAI(model="qwen3.7-plus", api_key=os.getenv("DASHSCOPE_API_KEY"),
#                  base_url=os.getenv("DASHSCOPE_API_BASE_URL"))

llm_qianwen = ChatTongyi(   
model="qwen-max",  
# api_key也可以不写，ChatTongyi组件内部默认回从env环境变量读取DASHSCOPE_API_KEY
api_key=os.getenv("DASHSCOPE_API_KEY"),
)

response = llm_qianwen.invoke("用一句话介绍下你自己")
print(response)

2.1.2 调用deepseek API

python
from langchain_deepseek import ChatDeepSeek
import dotenv
import os

dotenv.load_dotenv()

llm_deepseek = ChatDeepSeek(
    model="deepseek-v4-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    api_base=os.getenv("DEEPSEEK_BASE_URL"),
)

response = llm_deepseek.invoke("用一句话介绍下你自己")
print(response)

2.2 langchain 1.x 模型初始化

init_chat_model 是 LangChain 1.x 中推出的用于初始化聊天模型的统一接口。只要是LangChain支持的模型都可以处理，它会根据模型名称自动选择对应的模型类初始化实例。

基本语法:

python
from langchain.chat_models import init_chat_model

model = init_chat_model(
    "provider:model_name", # 提供商:模型名称
    api_key="your-api-key", # API 密钥（可选，可从环境变量读取）
    temperature=0.7, # 温度参数（可选）
    max_tokens=1000, # 最大 token 数（可选）
    **kwargs # 其他模型特定参数
)

2.2.1 通过init_chat_model初始化deepseek

python
from langchain.chat_models import init_chat_model
import dotenv
import os

dotenv.load_dotenv()
llm = init_chat_model(
    model="deepseek-v4-flash",
    model_provider="deepseek",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    api_base=os.getenv("DEEPSEEK_BASE_URL"),
)

response = llm.invoke("用一句话介绍下你自己")
print(response)

注意： model_provider可以省略，model和model_provider可以合起来写 eg: model= "deepseek:deepseek-v4-flash"

2.2.2 通过init_chat_model初始化 qianwen

由于langchain官方的模型提供商里没有阿里的，如果用阿里云百炼的模型会有点区别,


from langchain.chat_models import init_chat_model
import dotenv
import os

dotenv.load_dotenv()
llm = init_chat_model(
    model="qwen3.7-plus",
    model_provider="openai",  # 关键：指定使用openai兼容协议
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_API_BASE_URL")
)

response = llm.invoke("用一句话介绍下你自己")
print(response)

2.3 调用ollama本地模型

ChatOllama方式

python
from langchain_ollama import ChatOllama

model = ChatOllama(
    model= "deepseek-r1:7b",
    base_url="http://0.0.0.0:11434"
)
response = model.invoke("你是谁？")
print(response)

init_chat_model方式

python
from langchain.chat_models import init_chat_model

model = init_chat_model(
    model= "deepseek-r1:7b",
    model_provider="ollama",
    base_url="http://0.0.0.0:11434"
)
response = model.invoke("你是谁？")
print(response)

注意：本地如果开启代理的话，这里会出现502错误

3. model的调用

在 LangChain 中，模型调用（Invocation）是指通过特定方法触发大语言模型生成输出的过程。根据不同的应用场景和需求，LangChain 提供了几种核心的调用方式，主要是 invoke() 、 stream() 和 batch() 方法，以及它们的异步版本 ainvoke() 、 astream() 和 abatch() ，下面将系统地介绍这些方法。

invoke() ：阻塞式，一次性返回完整结果问答、批处理任务、无需实时反馈的场景。
ainvoke() ：非阻塞式，提高系统吞吐量高并发Web应用、IO密集型任务。
stream() ：流式输出，实时返回每个token聊天机器人、长文本生成、需要提升用户体验的交互应用。
asteam() ：非阻塞式，提高系统吞吐量高并发Web应用、IO密集型任务。
batch() ：批量处理多个输入高并发场景，需要同时处理大量请求。
abatch() ：非阻塞式，提高系统吞吐量高并发Web应用、IO密集型任务。

3.1 invoke()

invoke() 是 LangChain 中最核心的方法，它的工作模式是阻塞式的，即程序会等待模型完全生成整个响应后，再一次性将结果返回给用户。

基本语法

python
response = model.invoke(input, config=None)

参数	类型	说明	必须	默认值
input	`str` \| `list[dict]` \|`list[Message]` 等	你要发送给模型的内容	. 必须	无
config	`dict`	高级配置（回调函数、元数据、标签等）	可选	`None`

3.1.1 输入参数详解：

invoke方法非常灵活，支持三种形式的输入：文本输入、字典列表、消息对象列表。

文本输入这里就不过多介绍了，就是普通的字符串
字典列表(推荐，最灵活)

创建字典列表组成消息。一条消息通常包含 role（角色）、 content（内容）等信息。

✅适用场景：可以设置系统提示，表达多轮对话历史，JSON 兼容，易于序列化和网络传输，生产环境推荐。
❌缺点：代码稍微多一点（但更清晰）

python
messages = [
{"role": "system", "content": "系统提示"},
{"role": "user", "content": "用户消息"},
{"role": "assistant", "content": "AI回复"}, # 可选，用于对话历史
{"role": "user", "content": "继续提问"}
]

多轮对话，传递历史消息

python
conversation = [
    {"role": "system", "content": "你是一个非常友好的AI助手"},
    {"role": "user", "content": "你好，我叫小明"}
]
# 第一次对话
response1 = model.invoke(conversation)
# 打印响应
print(f"AI的回复1：{response1.content}")
# 添加记忆
conversation.append({"role": "assistant", "content": response1.content})
conversation.append({"role": "user", "content": "我叫什么名字？"})
# 第二次对话
response2 = model.invoke(conversation)
print(f"AI的回复2：{response2.content}")

消息对象列表使用内置的消息类（如 SystemMessage, HumanMessage, AIMessage），将消息对象列表输入模型。 ✅适用场景：需要类型检查（针对大型项目）、IDE 自动补全的场景 ❌缺点：代码较长、不如字典简洁、难以序列化（JSON）

python
from langchain_core.messages import SystemMessage, AIMessage, HumanMessage
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
# 从.env文件中加载环境变量
load_dotenv()
model = init_chat_model(
    model="qwen3.7-plus",
    model_provider="openai",  # 关键：指定使用openai兼容协议
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_API_BASE_URL")
)
# 使用消息对象格式构建消息
messages = [
SystemMessage("你是一个专业的数学老师。"),
HumanMessage("2 + 3 * 2 = ？"),
AIMessage("8"),
HumanMessage("我刚才问什么问题了？")
]
response = model.invoke(messages)
# 打印响应
print(f"AI的回复：{response.content}")

3.1.2 invoke返回值详解

python
from rich import print as rprint

# 使用字典格式构建消息
response = model.invoke([HumanMessage("2 + 3 * 2 = ？")])
# 打印响应
# print(type(response))
rprint(response)

nvoke 返回一个 AIMessage对象

python
AIMessage(
    # 模型生成的最终文本答案
    content='根据数学中的运算优先级（先乘除后加减）：\n\n1. 先计算乘法：3 * 2 = 
6\n2. 再计算加法：2 + 6 = 8\n\n所以，**2 + 3 * 2 = 8**。',
    additional_kwargs={'refusal': None},# 模型拒绝回答的情况（如触碰安全策略），None 表示正常回答
    
    # --- 响应元数据（API 返回的详细原始数据） ---
    response_metadata={
        'token_usage': {
            'completion_tokens': 340,  # 生成回答消耗的 Token 数（输出）
            'prompt_tokens': 20,       # 用户输入消耗的 Token 数（输入）
            'total_tokens': 360,       # 本次交互总共消耗的 Token
            'completion_tokens_details': {
                'accepted_prediction_tokens': None,  # 预测性生成的 Token 数
                'audio_tokens': None,                # 音频生成消耗（如有）
                'reasoning_tokens': 277,             # 推理模型（如 o1）思考过程消耗的 Token
                'rejected_prediction_tokens': None,    # 被拒绝的预测 Token
                'text_tokens': 340
            },
            'prompt_tokens_details': {
                'audio_tokens': None,      # 输入中的音频 Token 数
                'cached_tokens': 0,        # 命中的缓存 Token 数（能省钱/提速）
                'text_tokens': 20
            }
        },
        'model_provider': 'openai',
        'model_name': 'qwen3.7-plus',
        'system_fingerprint': None,  # 系统指纹，用于追踪模型后端的配置变更
        'id': 'chatcmpl-ce95d615-2d26-9348-8c79-23b28cc50af2',
        'finish_reason': 'stop',  # 停止原因：stop（自然结束）、length（长度受限）
        'logprobs': None
    },
    id='lc_run--019fcc8f-9fc5-7bb2-bea3-c661fca9f7fc-0',
    tool_calls=[],     # 正常触发的外部工具调用列表
    invalid_tool_calls=[],  # 触发失败或格式错误的工具调用
    
    # --- 统一消耗元数据（LangChain 标准化后的消耗格式） ---
    usage_metadata={
        'input_tokens': 20,   # 输入 Token 数
        'output_tokens': 340, # 输出 Token 数
        'total_tokens': 360,  # 总 Token 数
        'input_token_details': {'cache_read': 0}, # 从缓存中读取的输入数量
        'output_token_details': {'reasoning': 277}  # 包含在输出中的推理 Token
    }
)

3.2 stream 流式调用

invoke 和 stream 有什么区别？

invoke() ：同步调用，在模型输出完成后一次性获取响应，对于输出文本很长的场景，用户体验不好。
stream() ：流式调用，实时返回响应片段。调用后，返回一个迭代器(iterator) ，可以通过循环来实时处理每一个新生成的chunk内容块。

python
from langchain.chat_models import init_chat_model
import dotenv
import os

dotenv.load_dotenv()
model = init_chat_model(
    model="qwen3.7-plus",
    model_provider="openai",  # 关键：指定使用openai兼容协议
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_API_BASE_URL")
)
for chunk in model.stream("帮我解释下什么是人工智能"):
    print(chunk.text,end="",flush=True)

stream()方式的优点：

响应速度更快 — 用户不必等待完整输出
交互体验更流畅 — 尤其在长文本或复杂推理场景下
可实时展示模型思考过程

3.3 batch 批量调用

batch() 方法允许你一次性发送一组请求（含多条独立请求），模型会在后台并行处理，然后返回所有结果的列表。

与逐个顺序调用（invoke）相比，能大幅减少网络往返开销和等待时间，显著提升性能、降低成本。

适用场景：文档摘要、批量问答、数据预处理、多样本分类等。

按输入顺序一次性接收所有响应

batch()特点是等待所有请求处理完毕，按原始输入顺序返回结果列表。

python
from langchain.chat_models import init_chat_model
import dotenv
import os

dotenv.load_dotenv()
model = init_chat_model(
    model="qwen3.7-plus",
    model_provider="openai",  # 关键：指定使用openai兼容协议
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_API_BASE_URL")
)

messages = [
"你好，你是谁？",
"2 + 3 * 5 = ?",
"中国首都在哪里？"
]
responses = model.batch(messages)
for response in responses:
    print(response)

按完成顺序接收响应

当输入列表很大或单个模型调用耗时差异显著时， batch_as_completed() 允许应用在收到第一个结果后立即返回响应，而不会等待批次内所有任务完成才响应。即batch_as_completed() 每个请求完成后立即 yield 结果，结果可能乱序。

但是，每个返回的响应都被放在一个元组中，元组的第一个元素是原始输入的 index 索引，可根据索引重新排序。

python
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
# 从.env文件中加载环境变量
load_dotenv(override=True)
model = init_chat_model(
    model="qwen3.7-plus",
    model_provider="openai",  # 关键：指定使用openai兼容协议
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_API_BASE_URL")
)
messages = [
    "你好，你是谁？",
    "2 + 3 * 5 = ?",
    "中国首都在哪里？"
]
responses = model.batch_as_completed(messages)
for response in responses:
    print(response)

3.4 异步调用

在LangChain框架中，异步方法（ainvoke、astream、abatch）与它们的同步版本（invoke、 stream、batch）相比，具备如下特点：

避免阻塞主线程：同步调用会阻塞程序执行，而异步方法让应用程序在等待API响应时保持响应性。
优化资源利用：异步操作可以更高效地利用系统资源，减少空闲等待时间

python
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
import asyncio
import time
load_dotenv()
model = init_chat_model(
    model="qwen3.7-plus",
    model_provider="openai",  # 关键：指定使用openai兼容协议
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_API_BASE_URL")
)
async def demo_async_invoke():
    print("=== 演示：ainvoke 的异步（非阻塞）效果 ===")
    start_time = time.perf_counter() # 记录开始时间
    print("程序开始...")
    # 1. 创建任务 (Task)
    print(">>> 发起异步模型调用 (ainvoke)...")
    async_task = asyncio.create_task(model.ainvoke("用一句话解释人工智能。"))
    # 2. 并行执行其他任务
    print(">>> 模型请求已在后台发送，继续执行本地逻辑...")
    for i in range(3):
        await asyncio.sleep(1) # 使用异步等待，释放控制权
        print(f">>> 正在执行第{i + 1}个任务... (已耗时 {time.perf_counter() -
start_time:.2f}s)")
    # 3. 获取模型结果
    print(">>> 本地任务完成，检查模型状态...")
    response = await async_task
    end_time = time.perf_counter()
    print(f">>> 模型返回: {response.content}")
    print(f"=== 总运行耗时: {end_time - start_time:.2f}s ===")

async def main():
    """主函数"""
    await demo_async_invoke()

if __name__ == "__main__":
    asyncio.run(main())

输出如下：

plaintext
=== 演示：ainvoke 的异步（非阻塞）效果 ===
程序开始...
>>> 发起异步模型调用 (ainvoke)...
>>> 模型请求已在后台发送，继续执行本地逻辑...
>>> 正在执行第1个任务... (已耗时 1.00s)
>>> 正在执行第2个任务... (已耗时 2.00s)
>>> 正在执行第3个任务... (已耗时 3.00s)
>>> 本地任务完成，检查模型状态...
>>> 模型返回: 人工智能是一门让机器模拟人类智能，使其能够自主学习、推理、感知和解决问题的技术。
=== 总运行耗时: 6.10s ===

astream示例：

python
async def demo_async_stream():
    """演示异步调用的非阻塞特性"""
    print("=== 演示：astream 的异步（非阻塞）效果 ===")
    start_time = time.perf_counter() # 记录开始时间
    print("程序开始...")
    # 1. 发起异步流式请求
    # 注意：此时请求已发出，返回的是一个异步生成器
    print(">>> 发起异步流式调用 (astream)...")
    stream_resp = model.astream("请用一句话解释机器学习的基本概念。")
    # 2. 在等待流式响应的同时，执行其他任务
    print(">>> 流式请求已发送，程序无需等待，继续执行其他异步任务...")
    for i in range(3):
        # 使用 asyncio.sleep 而非 time.sleep
        # 这允许事件循环在等待时去处理上面的 stream_resp 网络 IO
        await asyncio.sleep(1)
        #    print(f">>> 正在执行并发任务 {i + 1}... ")
        print(f">>> 正在执行第{i + 1}个任务... (已耗时 {time.perf_counter() - start_time:.2f}s)")
    # 3. 现在开始处理流式结果
    print(">>> 模拟任务已完成，开始读取缓冲区中的流式结果...")
    end_time = time.perf_counter()
    print(">>> 流式输出: ", end="", flush=True)
    async for chunk in stream_resp:
        # LangChain 的消息块通常通过 .content 获取内容
        content = chunk.content if hasattr(chunk, 'content') else str(chunk)
        print(content, end="", flush=True)
    print("\n>>> 流式输出结束\n")
    print(f"=== 总运行耗时: {end_time - start_time:.2f}s ===")

输出：

plaintext
=== 演示：astream 的异步（非阻塞）效果 ===
程序开始...
>>> 发起异步流式调用 (astream)...
>>> 流式请求已发送，程序无需等待，继续执行其他异步任务...
>>> 正在执行第1个任务... (已耗时 1.00s)
>>> 正在执行第2个任务... (已耗时 2.00s)
>>> 正在执行第3个任务... (已耗时 3.00s)
>>> 模拟任务已完成，开始读取缓冲区中的流式结果...
>>> 流式输出: 机器学习是一种让计算机通过从数据中自动学习规律，从而在没有明确编程指令的情况下，凭借经验不断提升预测和决策能力的技术。
>>> 流式输出结束

=== 总运行耗时: 3.00s ===

async_batch示例：

python
async def demo_async_batch():
    """演示异步批量的非阻塞特性"""
    print("=== 演示：abatch 的异步（非阻塞）效果 ===")
    start_time = time.perf_counter() # 记录开始时间

    print("程序开始...")

    # 准备批量输入
    questions = ["用一句话说明深度学习与传统机器学习的区别", "中国首都在哪里？"]
    # 1. 发起异步批量请求
    # 关键修改：使用 create_task 让协程立即在后台执行
    print(">>> 发起异步批量调用 (abatch)...")
    batch_task = asyncio.create_task(model.abatch(questions))
    # 2. 在等待批量处理的同时，执行其他任务
    print(">>> 批量任务已在后台运行，主程序继续执行...")
    for i in range(3):
        # 关键修改：使用 asyncio.sleep 允许后台任务获取 CPU 时间片进行网络请求
        await asyncio.sleep(1)
        print(f">>> 正在执行第{i + 1}个任务... (已耗时 {time.perf_counter() - start_time:.2f}s)")

    # 3. 等待批量处理结果
    print(">>> 其他任务已完成，现在获取后台批量任务的结果...")
    # 此时 batch_task 可能已经完成，或者我们在这里等待它完成
    responses = await batch_task

    end_time = time.perf_counter()

    for response in responses:
        content = response.content if hasattr(response, 'content') else str(response)
        print(f">>> 响应内容: {content}")

    print(f"=== 总运行耗时: {end_time - start_time:.2f}s ===")

输出如下：

plaintext
=== 演示：abatch 的异步（非阻塞）效果 ===
程序开始...
>>> 发起异步批量调用 (abatch)...
>>> 批量任务已在后台运行，主程序继续执行...
>>> 正在执行第1个任务... (已耗时 1.00s)
>>> 正在执行第2个任务... (已耗时 2.00s)
>>> 正在执行第3个任务... (已耗时 3.00s)
>>> 其他任务已完成，现在获取后台批量任务的结果...
>>> 响应内容: 传统机器学习依赖人工设计和提取特征，而深度学习通过多层神经网络自动从原始数据中进行端到端的特征提取与学习。
>>> 响应内容: 中国的首都是北京。
=== 总运行耗时: 8.87s ===

扩展

模型配置信息profile

LangChain1.1及更高版本可以通过 profile属性查看模型的配置信息。这是LangChain针对模型的能力画像，但是否存在，取决于LangChain在集成模型厂商的服务时是否声明了能力画像。

python
from langchain_openrouter import ChatOpenRouter
from dotenv import load_dotenv
from rich import print as rprint
# 从.env文件中加载环境变量
load_dotenv(override=True)
model = ChatOpenRouter(
    model="openai/gpt-4o-mini",
    # model="deepseek/deepseek-v3.2",
    temperature=0.7,
    timeout=30,
    max_tokens=1000,
    max_retries=6
)
rprint(model.profile)

输出如下：

plaintext
{
    'max_input_tokens': 128000,
    'max_output_tokens': 16384,
    'text_inputs': True,
    'image_inputs': True,
    'audio_inputs': False,
    'video_inputs': False,
    'text_outputs': True,
    'image_outputs': False,
    'audio_outputs': False,
    'video_outputs': False,
    'reasoning_output': False,
    'tool_calling': True,
    'structured_output': True
}

目录

1. 环境搭建

2. model 的创建

2.1 调用模型提供商API

2.2 langchain 1.x 模型初始化

2.3 调用ollama本地模型

3. model的调用

3.1 invoke()

3.1.1 输入参数详解：

3.1.2 invoke返回值详解

3.2 stream 流式调用

3.3 batch 批量调用

3.4 异步调用

扩展

模型配置信息profile