LangChain1.2学习第二章—— 调用大模型

# =========================================================
# LangChain 核心框架
# 作用：LangChain 1.x 主体、核心抽象、社区组件、实验性组件、文本切分器
# =========================================================
langchain==1.2.12
langchain-core==1.2.18
langchain-community==0.4.1
langchain-classic==1.0.2
langchain-text-splitters==1.1.1
langchain-experimental==0.4.1

# =========================================================
# jupyter
# 作用：交互式编程记事本，默认安装最新版即可
# =========================================================
jupyter


# =========================================================
# ollama
# 作用：调用本地大模型
# =========================================================
langchain-ollama==1.0.1
ollama==0.6.2

# =========================================================
# LangGraph / Agent 编排
# 作用：构建 Agent、状态图、多步骤工作流、检查点、PostgreSQL 持久化
# =========================================================
langgraph==1.1.2
langgraph-prebuilt==1.0.8
langgraph-checkpoint==4.0.1
langgraph-checkpoint-postgres==3.0.5
langgraph-sdk==0.3.9


# =========================================================
# MCP / FastMCP 集成
# 作用：构建 MCP Server、连接 MCP 工具、资源、Prompt
# =========================================================
mcp==1.27.0
fastmcp==3.2.4
langchain-mcp-adapters==0.2.1


# =========================================================
# 大模型供应商与 LangChain 适配器
# 作用：接入 OpenAI 协议、DeepSeek、Anthropic、OpenRouter、通义千问、腾讯等模型
# =========================================================
openai==2.26.0
anthropic==0.84.0
langchain-openai==1.1.11
langchain-deepseek==1.0.1
langchain-anthropic==1.3.4
langchain-openrouter==0.1.0
openrouter==0.7.11
dashscope==1.25.6
tencentcloud-sdk-python==3.1.86
PyJWT==2.10.1


# =========================================================
# 搜索工具 / 外部工具集成
# 作用：接入 Tavily 等联网搜索工具
# =========================================================
langchain-tavily==0.2.17


# =========================================================
# Web 服务 / API / SSE
# 作用：MCP HTTP 服务、FastAPI 服务、SSE 流式通信、本地 API 服务
# =========================================================
fastapi==0.135.1
uvicorn==0.46.0
sse-starlette==3.3.4
httpx==0.28.1
httpx-sse==0.4.3
aiohttp==3.12.14
requests==2.32.5
requests-toolbelt==1.0.0
websockets==16.0
watchfiles==1.1.1


# =========================================================
# 配置管理 / 数据校验 / 序列化
# 作用：读取 .env、Pydantic 配置、JSON/YAML、结构化输出
# =========================================================
python-dotenv==1.2.1
pydantic==2.12.5
pydantic-settings==2.12.0
PyYAML==6.0.3
orjson==3.11.7
jsonschema==4.26.0
jsonref==1.1.0
dataclasses-json==0.6.7


# =========================================================
# 日志 / 命令行 / 调试辅助
# 作用：日志输出、CLI、富文本终端输出、重试机制
# =========================================================
loguru==0.7.3
rich==14.3.3
typer==0.24.1
click==8.3.1
tenacity==9.1.4
tqdm==4.67.3
python-dateutil==2.9.0.post0
pytz==2026.2


# =========================================================
# 测试工具
# 作用：单元测试、教程代码验证
# =========================================================
pytest==9.0.3

# =========================================================
# 记忆相关
# 作用：PostgreSQL 检查点、SQL 数据源
# =========================================================
psycopg[binary]==3.3.3
psycopg-pool==3.3.0

# ===========如下注释部分，在RAG章节才需要安装，暂时注释掉==============

# =========================================================
# RAG / 向量数据库 / 数据库连接
# 作用：Milvus 向量库、PostgreSQL 检查点、SQL 数据源
# =========================================================
# langchain-milvus==0.3.3
# pymilvus==2.6.12
# SQLAlchemy==2.0.48


# =========================================================
# Embedding / Tokenizer / 文本处理
# 作用：TokenTextSplitter、SemanticChunker、文本相似度、传统 NLP 处理
# 注意：这里不包含 sentence-transformers，也不包含 torch
# =========================================================
# tiktoken==0.12.0
# numpy==2.4.4
# scipy==1.17.1
# scikit-learn==1.8.0
# nltk==3.9.4
# regex==2026.2.28
# langdetect==1.0.9


# =========================================================
# HuggingFace / Transformers 基础组件
# 作用：本地模型、Tokenizer、部分文档解析模型可能会用到
# 注意：不包含 torch；如果加载本地深度学习模型，请单独安装匹配 CUDA 的 PyTorch
# =========================================================
# transformers==5.3.0
# tokenizers==0.22.2
# huggingface-hub==1.11.0
# safetensors==0.7.0

# =========================================================
# PyTorch的安装--cpu版本
# 作用：Unstructured的依赖
# =========================================================
# torch==2.11.0
# torchvision==0.26.0

# =========================================================
# Unstructured / LangChain Loader 文档解析
# 作用：PDF、Word、PPT、Excel、HTML、Markdown、图片文档等 Loader 支持
# 注意：unstructured-inference 可能依赖本地推理环境，torch/torchvision 请单独安装
# =========================================================
# unstructured==0.20.6
# unstructured-client==0.44.0
# unstructured-inference==1.6.11
# unstructured.pytesseract==0.3.15
# 
# pdfminer.six==20260107
# pdf2image==1.17.0
# pypdf==6.10.2
# pypdfium2==5.8.0
# pikepdf==10.5.1
# pi-heif==1.3.0
# pillow==12.2.0
# opencv-python==4.13.0.92
# onnx==1.21.0
# onnxruntime==1.25.1
# 
# python-docx==1.2.0
# python-pptx==1.0.2
# openpyxl==3.1.5
# xlrd==2.0.2
# xlsxwriter==3.2.9
# pandas==3.0.2
# 
# beautifulsoup4==4.14.3
# html5lib==1.1
# lxml==6.1.0
# Markdown==3.10.2
# jq==1.11.0
# filetype==1.2.0
# python-magic==0.4.27
# python-iso639==2026.4.20
# msoffcrypto-tool==6.0.0
# python-oxmsg==0.0.2
# olefile==0.47
# pypandoc-binary==1.17

二、模型初始化

2.1 使用厂商提供的API

在 LangChain 中初始化模型，主要可以通过直接使用特定的Model Class和使用统一的 init_chat_model函数这两种方式来实现。

这里先讲方式1，这种方式最直接。LangChain为一些大模型供应商提供了专门的Model类，导入对应的具体类（如 ChatOpenAI、ChatAnthropic、ChatDeepSeek、ChatOllama、ChatHunyuan、 ChatTongyi、ChatZhipuAI）并进行实例化。

以下使用DeepSeek举例

步骤1：配置.env文件（明确去deepseek官网获取key)

# DeepSeek
DEEPSEEK_API_KEY=xxx
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_MODEL=deepseek-v4-flash

#阿里QWEN
QWEN_API_KEY=xxx
QWEN_BASE_URL=xxx
QWEN_MODEL=qwen3.7-flash

步骤2：读取配置并初始化模型

import os

from dotenv import load_dotenv
from langchain_deepseek import ChatDeepSeek

#1、读取env参数信息。
# 使用load_dotenv()方法加载.env中配置的参数
# override = True：无论操作系统、终端、虚拟环境是否已经存在同名的环境变量，都会强势使用.env配置的内容覆盖
load_dotenv(override=True)
#读取配置信息
DEEPSEEK_API = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")
DEEPSEEK_MODEL = os.getenv("DEEPSEEK_MODEL")

#2、模型初始化
llm_deepseek = ChatDeepSeek(
    model=DEEPSEEK_MODEL,
    api_key=DEEPSEEK_API,
    api_base = DEEPSEEK_BASE_URL,
)
#3、调用大模型
response = llm_deepseek.invoke("你是谁")
print(response)

输出如下：

content='你好！我是 DeepSeek，由深度求索公司创造的 AI 助手。我可以帮你解答问题、处理文本、分析文件等。有什么可以帮你的吗？' additional_kwargs={'refusal': None, 'reasoning_content': 'We need answer in Chinese likely. Need introduce AI assistant. Need maybe "我是DeepSeek" etc. Need follow. User asks "你是谁" = Who are you. We answer. Need mention model? We are DeepSeek latest. Keep concise.'} response_metadata={'token_usage': {'completion_tokens': 88, 'prompt_tokens': 84, 'total_tokens': 172, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 52, 'rejected_prediction_tokens': None, 'text_tokens': None}, 'prompt_tokens_details': {'audio_tokens': None, 'cache_write_tokens': None, 'cached_tokens': 0, 'image_tokens': None, 'text_tokens': None}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 84}, 'model_provider': 'deepseek', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'a26a7955944dc5c60445bff77fac9c8e', 'id': 'cd4282af-dc72-43fa-85fd-8034edb74c23', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--01a027a2-c255-7470-ab0b-d3d6d71e268f-0' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 84, 'output_tokens': 88, 'total_tokens': 172, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 52}}

注意：通过查看源码(chat_model.py)得知，子层会自动调用配置文件DEEPSEEK_API_KEY和DEEPSEEK_API_BASE。如果配置文件里的命名如上，那么在初始化大模型时，这两个参数可以省略。llm_deepseek = ChatDeepSeek( model=DEEPSEEK_MODEL, )，只需提供model_name即可。

2.2 兼容模式

LangChain没有为所有大模型厂商提供专用接口，见Langchain大模型集成列表。如果选用的平台没有专用接口，可以通过兼容接口调用。

另外专用接口的对接方式五花八门，如腾讯混元的ChatHunyuan需要单独的APP_ID + SecretId + SecretKey，配置繁琐，用户不友好。

结论：大多数API平台都支持OpenAI API接口规范，所以基本都可以通过 ChatOpenAI 集成。

from langchain_openai import ChatOpenAI
load_dotenv(override=True)
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")
DEEPSEEK_MODEL = os.getenv("DEEPSEEK_MODEL")
model = ChatOpenAI(
    model=DEEPSEEK_MODEL,
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
)
print(model.invoke("请简单介绍尔朱荣的军事能力"))

输出如下;

content='尔朱荣是北魏末期的军事统帅、权臣，军事能力非常突出，属于“枭雄级”将领。他的军事特点可概括为：\n\n1. **善于骑兵作战，敢打硬仗** \n 他长期坐镇北方，手握强悍的契胡骑兵，尤其擅长突击和奔袭，机动性强，攻击果断。\n\n2. **以少胜多的经典战例** \n 最典型的是**击败葛荣起义军**。当时葛荣号称百万之众，尔朱荣只带七千精骑，却利用地形和突袭战术，直冲敌阵，一举击溃葛荣主力，并临阵生擒葛荣，堪称军事奇迹。\n\n3. **战术灵活，重视破敌核心** \n 他的打法往往不是消耗战，而是集中精锐骑兵猛攻对方指挥中枢，靠“斩首战术”瓦解敌方阵线，再乘势追杀扩大战果。\n\n4. **平定河北、山东等地叛乱** \n 除了葛荣，他还先后击败邢杲、韩楼等势力，基本平定了北方的各路反魏力量，为北魏晚期延续统治立下“战绩”。\n\n5. **收降纳叛，却也很残暴** \n 他降服了高欢、侯景等后来叱咤风云的人物，说明他识人、能驾驭猛将。但他政治手腕粗野，发动“河阴之变”残杀朝臣，导致名声很坏，最终也被北魏孝庄帝设计杀死。\n\n总的来说，**尔朱荣是一流的战术型将领，擅长骑兵突击、以少打多，但缺乏政治智慧和治理天下的格局**。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 838, 'prompt_tokens': 92, 'total_tokens': 930, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 489, 'rejected_prediction_tokens': None, 'text_tokens': None}, 'prompt_tokens_details': {'audio_tokens': None, 'cache_write_tokens': None, 'cached_tokens': 0, 'image_tokens': None, 'text_tokens': None}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 92}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'a26a7955944dc5c60445bff77fac9c8e', 'id': '4d2510dc-c430-46e9-970b-1ee8ea590fc4', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--01a027ad-3fe2-7962-9996-f478410ffe83-0' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 92, 'output_tokens': 838, 'total_tokens': 930, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 489}}

2.3 使用LangChain提供的统一写法

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

from langchain.chat_models import init_chat_model
import os
from dotenv import load_dotenv

#1.加载配置文件,获取参数
load_dotenv(override=True)
DEEPSEEK_API = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")
DEEPSEEK_MODEL = os.getenv("DEEPSEEK_MODEL")

#2.大模型初始化
model = init_chat_model(
    model=DEEPSEEK_MODEL,
    model_provider="deepseek",
    api_key=DEEPSEEK_API,
    base_url=DEEPSEEK_BASE_URL,
)
#3.使用大模型
print(model.invoke("请简单介绍破六韩拔陵").content)

输出如下：

破六韩拔陵（？—525），北魏末年“六镇起义”的首领。“破六韩”是复姓（又作“破六汗”），“拔陵”是名，出身匈奴族。

北魏正光四年（523年），因为六镇镇兵、镇民长期受压迫，又遇饥荒，他在沃野镇（今内蒙古境内）杀镇将起兵，自称“真王”。起义迅速得到北方六镇响应，声势浩大，史称“六镇起义”。

北魏朝廷镇压不利，于是联合柔然。孝昌元年（525年），柔然可汗阿那瑰出兵大败起义军，破六韩拔陵兵败被杀。

这场起义动摇了北魏的统治，成为北魏分裂为东魏、西魏的重要导火索。

问题： init_chat_model 和直接使用 ChatTongyi、ChatOpenAI、ChatDeepSeek有什么区别？

回答： init_chat_model 是 LangChain 1.0 的统一接口，优势包括：

统一接口：无需记住每个提供商的不同初始化方式（以一致的方式初始化）

易于切换：简化了智能体系统中模型切换策略（只需修改模型字符串）

简洁明了：更简洁的语法，减少样板代码

自动适配：内部根据模型标识自动选择对应的驱动类(ChatOpenAI、ChatDeepSeek）

问题1：model_provider支持哪些provider？

model_provider表示模型的提供者，支持的providers有：anthropic , anthropic_bedrock, azure_ai, azure_openai, bedrockbedrock_converse, cohere, deepseek , fireworks, google_anthropic_vertex, google_genai, google_vertexaigrog, huggingface, ibm, mistralai, nvidia, ollama, openai , openrouter , perplexity, together, upstage, xai。

如果 model_provider="openai"，会自动加载langchain-openai的依赖包，底层调用的是 ChatOpenAI 类。
如果 model_provider="deepseek"，会自动加载langchain-deepseek的依赖包，底层调用的是ChatDeepSeek类。
像阿里的dashscope尚未被LangChain官方纳入模型的统一注册体系，暂时不知道"dashscope"的提供者是谁。此时可以将model_provider设置为openai，底层将会用openai的规范处理请求，这就要求我们调用的模型服务是OpenAI Compatible的。

问题2：如果在model参数中没有指明模型提供者，必须在model_provider中指明？

可以在model参数中通过前缀指定模型供应商，和模型名称之间用冒号分割，等价于通过。 model_provider参数指定供应商。如果两个位置都没有指明供应商，LangChain底层会按照内置规则自动推断。

但是，并非所有的模型都支持自动推断，如model名称qwen-plus不支持自动推断，没有指明供应商会报错。

2.4 总结

DeepSeek官网的DeepSeek模型：可以调用ChatDeepSeek()、ChatOpenAI()、 init_chat_model()三种方式

阿里云百炼平台的DeepSeek模型：可以调用ChatTongyi()、ChatOpenAI()、init_chat_model() 三种方式

OpenRouter平台的DeepSeek模型：可以调用ChatOpenRouter()、ChatOpenAI()、 init_chat_model()三种方式

CloseAPI平台的DeepSeek模型：可以调用ChatOpenAI()、init_chat_model() 两种方式

2.5 参数介绍

1、temperature 参数根据使用场景选择：

0.0-0.3：需要一致性、准确性的任务（数学计算、数据提取、分类、代码生成）

0.5-0.7：平衡创造性和一致性（聊天、问答）

0.8-1.5：创造性任务（写作、头脑风暴）

1.5-2.0：高度创造性（诗歌、故事创作）

2、Token是什么？

基本单位: 大模型通过分词器（Tokenizer）将文本拆分后的最小语义单元是token（相当于自然语言中的词或字）。不同的模型采用不同的分词算法（如BPE、WordPiece），因此同一段文本在不同模型中的Token数量可能不同。

收费依据：大语言模型通常也是以token的数量作为其计量（或收费）的依据。

1个中文Token≈1-1.8个汉字，1个英文Token≈3-4个字符
Token与字符转化的可视化工具：
- OpenAI提供：https://platform.openai.com/tokenizer
- 百度智能云提供：https://console.bce.baidu.com/support/#/tokenizer

三、本地部署大模型

3.1 Ollama的介绍

LangChain也支持使用 Ollama 、 vLLM 等框架启动的本地大模型。这里以Ollama为例进行演示。 Ollama是在Github上的一个开源项目，其项目定位是：一个本地运行大模型的集成框架，可以实现如 Qwen、Deepseek 等主流大模型的下载、启动和本地运行的自动化部署及推理流程。

Ollama官方地址： https://ollama.com

产品定位：

ollama安装、本地部署大模型请自行百度

3.2 调用本地大模型

方式一：使用ChatOllama调用

from langchain_ollama import ChatOllama
ollama_llm = ChatOllama(
   model="deepseek-r1:7b-qwen-distill-q4_K_M",
    base_url="http://localhost:11434",
)
print(ollama_llm.invoke("你是谁").content)

输出如下

您好！我是由中国的深度求索（DeepSeek）公司开发的智能助手DeepSeek-R1。如您有任何任何问题，我会尽我所能为您提供帮助。

方式二：使用init_chat_model调用

from langchain.chat_models import init_chat_model
ollama_llm = init_chat_model(
    model="deepseek-r1:7b-qwen-distill-q4_K_M",
    model_provider="ollama"
)
print(ollama_llm.invoke("deepseek是什么").content)

输出如下：

深度求索人工智能基础技术研究有限公司(简称“深度求索”或“DeepSeek”)，成立于2023年，是一家专注于实现AGI的中国公司。

四、模型调用

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

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

4.1 invoke()

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

4.1.1 invoke()说明

简单来说，invoke 方法的作用就是：

接收用户的输入（问题、指令、对话历史等）
发送给大模型（如 GPT-4、Llama、Claude 等)
返回模型的响应（文本回复 + 元数据信息）

基本语法：

response = model.invoke(input, config=None）

参数详解

4.1.2 输入参数详解

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

1、文本输入

简单的一次性问答，直接传入一个问题或指令。

✅适用场景：快速测试，不需要保留对话历史的简单生成任务。

❌缺点：无法设置系统提示（system prompt），无法传递对话历史

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

from openai import conversations

#从 .env 中加载配置参数
load_dotenv(override=True)
qwen_model = init_chat_model(
    model="deepseek:deepseek-v4-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
)
print(qwen_model.invoke("请介绍宇文泰").content)

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

✅适用场景：可以设置系统提示，表达多轮对话历史，JSON 兼容，易于序列化和网络传输，生产环境推荐。

❌缺点：代码稍微多一点（但更清晰）

from langchain_qwq import ChatQwen
from dotenv import load_dotenv
import os
load_dotenv(override=True)
qwen_model = ChatQwen(
    model="qwen-plus",
    api_key=os.getenv("QWEN_API_KEY"),
    base_url=os.getenv("QWEN_BASE_URL"),
)
# 使用字典格式构建消息
messages = [
    {"role":"system","content":"你是一位专业的历史学家，尤其擅长南北朝历史。"},
    {"role":"user","content":"请简单介绍高欢"}
]
print(f"AI的回复是:{qwen_model.invoke(messages).content}")

角色说明

多轮对话，包含历史

from langchain_qwq import ChatQwen
from dotenv import load_dotenv
import os
load_dotenv(override=True)
qwen_model = ChatQwen(
    model="qwen-plus",
    api_key=os.getenv("QWEN_API_KEY"),
    base_url=os.getenv("QWEN_BASE_URL"),
)
messages = [
    {"role":"system","content":"你是一位小学数学老师。"},
    {"role":"user","content":"2 + 3 = ?"},
    {"role":"assistant","content":"5"},
    {"role":"user","content":"我刚才问了什么问题"},
]
print(f"AI的回复是:{qwen_model.invoke(messages).content}")

如果不传递历史，AI 会"失忆"

from langchain_qwq import ChatQwen
from dotenv import load_dotenv
import os
load_dotenv(override=True)
qwen_model = ChatQwen(
    model="qwen-plus",
    api_key=os.getenv("QWEN_API_KEY"),
    base_url=os.getenv("QWEN_BASE_URL"),
)
messages = [
    {"role":"system","content":"你是大模型"},
    {"role":"user","content":"2 + 3 = ?"}
]
print(f"AI第一次回复是:{qwen_model.invoke(messages).content}")
message1 = [
    {"role":"user","content":"我刚才的问题是什么"}
]
print(f"AI第二次回复是:{qwen_model.invoke(message1).content}")

作为对比，传递记忆

from langchain_qwq import ChatQwen
from dotenv import load_dotenv
import os
load_dotenv(override=True)
qwen_model = ChatQwen(
    model="qwen-plus",
    api_key=os.getenv("QWEN_API_KEY"),
    base_url=os.getenv("QWEN_BASE_URL")
)
messages = [
    {"role":"system","content":"你是大模型"},
    {"role":"user","content":"2 + 3 = ?"}
]
print(f"AI第一次回复是:{qwen_model.invoke(messages).content}")

#添加记忆
messages.append({"role":"assistant","content":qwen_model.invoke(messages).content})
messages.append({"role":"user","content":"我刚才的问题是什么"})
print(f"AI第二次回复是:{qwen_model.invoke(messages).content}")

4.1.3 大模型返回值介绍

invoke方法返回一个AIMessage对象，源码如下

    def invoke(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> AIMessage:
        config = ensure_config(config)
        return cast(
            "AIMessage",
            cast(
                "ChatGeneration",
                self.generate_prompt(
                    [self._convert_input(input)],
                    stop=stop,
                    callbacks=config.get("callbacks"),
                    tags=config.get("tags"),
                    metadata=config.get("metadata"),
                    run_name=config.get("run_name"),
                    run_id=config.pop("run_id", None),
                    **kwargs,
                ).generations[0][0],
            ).message,
        )

AIMessage中包含丰富的信息，假设输出为例，整体说明如下

AIMessage(
    # --- 核心内容 ---
content='2 + 3 * 2 = **8**',  # 模型生成的最终文本答案
    
    additional_kwargs={
        'refusal': None# 模型拒绝回答的情况（如触碰安全策略），None 表示正常回答
    },
    
    # --- 响应元数据（API 返回的详细原始数据） --
    response_metadata={
        'token_usage': {
            'completion_tokens': 15,    # 生成回答消耗的 Token 数（输出）
            'prompt_tokens': 16,        # 用户输入消耗的 Token 数（输入）
            'total_tokens': 31,         # 本次交互总共消耗的 Token
            
            'completion_tokens_details': {
                'accepted_prediction_tokens': 0, # 预测性生成的 Token 数
                'audio_tokens': 0,               # 音频生成消耗（如有）
                'reasoning_tokens': 0,# 推理模型（如 o1）思考过程消耗的 Token
                'rejected_prediction_tokens': 0  # 被拒绝的预测 Token
            },
            
            'prompt_tokens_details': {
                'audio_tokens': 0,   # 输入中的音频 Token 数
                'cached_tokens': 0   # 命中的缓存 Token 数（能省钱/提速）
            },
            
            # --- 延迟性能监控（单位：毫秒 ms） --
            'latency_checkpoint': {
                'engine_tbt_ms': 4,        # 引擎 Token 间平均间隔时间
                'engine_ttft_ms': 36,      # 引擎生成首个 Token 的时间
                'engine_ttlt_ms': 100,     # 引擎生成最后一个 Token 的时间
                'pre_inference_ms': 86, # 推理前的预处理耗时(安全审核、Token 化等
预处理)
                'service_tbt_ms': 4, # 服务端token与token之间生成的间隔时间，决定
了打字机效果是否丝滑。
                'service_ttft_ms': 280,  # 服务端接收到请求到输出首字的总时间
                'service_ttlt_ms': 338,    # 服务端完成全部输出的总时间
                'total_duration_ms': 259,  # 本次请求在系统中记录的总持续时长
                'user_visible_ttft_ms': 194   # 用户看到第一个字跳出来等待的时间
            }
        },
        'model_provider': 'openai',               # 模型供应商
        'model_name': 'gpt-5.4-mini-2026-03-17',  # 使用的具体模型版本
        'system_fingerprint': None,          # 系统指纹，用于追踪模型后端的配置变
更
        'id': 'chatcmpl-DgWobsxhDOqzjqVFwbZYKRnovpEiV', #API层面的响应 ID
        'service_tier': 'default',    # 服务层级（如按量付费或订阅）
        'finish_reason': 'stop',     # 停止原因：stop（自然结束）、length（长度受
限）
        'logprobs': None            # 对数概率（通常用于分析词汇选择的可能性）
    },
    
    # --- LangChain 内部标识 --
    id='lc_run--019e3659-5ee2-7b62-bc8a-741e27374b43-0', 
    # LangChain 追踪此条运行的唯一 ID
    
    # --- 工具调用信息 --
    tool_calls=[],             # 正常触发的外部工具调用列表
    invalid_tool_calls=[],     # 触发失败或格式错误的工具调用
# --- 统一消耗元数据（LangChain 标准化后的消耗格式） --
usage_metadata={
'input_tokens': 16,    
# 输入 Token 数
'output_tokens': 15,   # 输出 Token 数
'total_tokens': 31,    
# 总 Token 数
'input_token_details': {
'audio': 0, 
'cache_read': 0    
# 从缓存中读取的输入数量
},
'output_token_details': {
'audio': 0, 
'reasoning': 0     
# 包含在输出中的推理 Token
}
}
)

总结一下:

1. 核心内容与基本信息：

content : 模型生成的文本回答。这是你最关心的核心输出。
id : 本次运行在 LangChain 内部生成的唯一标识符（Run ID）。
additional_kwargs : 包含特定供应商的额外参数。
- refusal : 如果模型拒绝回答（涉及敏感政策），此处会显示拒绝原因。

2.消耗统计 (Token Usage)——这部分决定了你这一行输入操作花了多少钱：

prompt_tokens / input_tokens : 输入 Token 数。你发送给模型的问题长度。
completion_tokens / output_tokens : 输出 Token 数。模型回答生成的长度。
total_tokens : 总消耗。两者之和。
reasoning_tokens : 推理 Token 数。如果是 O1/O3 等推理模型，这里会显示它在“思考”时消耗的 Token。
cached_tokens : 缓存命中的 Token 数。重复提问时，如果命中了模型商的缓存，这部分费用通常更低。

3.响应元数据 (Response Metadata)——部分是 API 返回的原始详细信息：

model_name : 实际调用的模型具体版本.
model_provider : 模型供应商
finish_reason : 生成停止的原因。
- stop : 正常回答结束。
- length : 达到最大 Token 限制被截断。
system_fingerprint : 系统指纹，用于追踪模型后端的配置变更

4.性能与延迟 (Latency Checkpoint)——这是针对 API 响应速度的深度拆解（单位通常为毫秒 ms）

total_duration_ms : 总耗时。从请求发出到完全收到的总时间（259ms）
user_visible_ttft_ms : 首字到达时间。用户看到第一个字跳出来等待的时间（194ms），这是体感快慢的关键。
engine_ttft_ms : 引擎层面的首字到达时间（36ms）。
engine_ttlt_ms : 引擎生成最后一个字的时间（100ms）
pre_inference_ms : 推理前处理耗时。包括安全审核、Token 化等预处理（86ms）。
service_tbt_ms : Time Between Tokens。字与字之间生成的间隔时间，决定了打字机效果是否丝滑。

5.工具调用信息

tool_calls : 结构化工具调用列表。如果模型决定调用某个 Python 函数或搜索工具，参数会在这里。
invalid_tool_calls : 格式错误的工具调用尝试。

以下示例输出部分信息

from langchain_qwq import ChatQwen
from dotenv import load_dotenv
import os
load_dotenv(override=True)
qwen_model = ChatQwen(
    model="qwen-plus",
    api_key=os.getenv("QWEN_API_KEY"),
    base_url=os.getenv("QWEN_BASE_URL")
)
response = qwen_model.invoke("用一句话解释AI")
print(f"大模型回复:{response.content}")
#1.获取元数据
metadata = response.response_metadata
print(f"使用的模型: {metadata['model_name']}")
print(f"结束原因: {metadata['finish_reason']}")
print(f"模型提供商：{metadata['model_provider']}\n")
# 3. 获取 Token 使用情况
usage = metadata.get('token_usage', {})
print(f"输入 tokens: {usage.get('prompt_tokens')}")
print(f"输出 tokens: {usage.get('completion_tokens')}")
print(f"总计 tokens: {usage.get('total_tokens')}")
# 4. 获取消息 ID
print(f"消息 ID: {response.id}")

运行如下

4.2 流式调用

invoke 和 stream 有什么区别？

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

from langchain_qwq import ChatQwen
from dotenv import load_dotenv
import os
load_dotenv(override=True)
qwen_model = ChatQwen(
    model="qwen-plus",
    api_key=os.getenv("QWEN_API_KEY"),
    base_url=os.getenv("QWEN_BASE_URL")
)
for chunk in qwen_model.stream("写一首赞颂卫青的七言律诗"):
    print(chunk.text,end="",flush=True)

输出不再是整段返回，而是流式输出。

stream()方式的优点：

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

4.3 批量调用

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

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

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

4.3.1 一次性接收所有响应

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

from langchain_qwq import ChatQwen
from dotenv import load_dotenv
import os
load_dotenv(override=True)
qwen_model = ChatQwen(
    model="qwen-plus",
    api_key=os.getenv("QWEN_API_KEY"),
    base_url=os.getenv("QWEN_BASE_URL")
)
messages = [
    "请介绍侯莫陈崇",
    "在平定安史之乱中，谁的功能最大",
    "秦始皇的功与过"
]
response = qwen_model.batch(messages)
for result in response:
    print(f" {result.content}")

4.3.2 按完成顺序接收响应

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

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

from langchain_qwq import ChatQwen
from dotenv import load_dotenv
import os
load_dotenv(override=True)
qwen_model = ChatQwen(
    model="qwen-plus",
    api_key=os.getenv("QWEN_API_KEY"),
    base_url=os.getenv("QWEN_BASE_URL")
)
messages = [
    "请介绍侯莫陈崇",
    "在平定安史之乱中，谁的功能最大",
    "秦始皇的功与过"
]
response = qwen_model.batch_as_completed(messages)
for result in response:
    print(f" {result}")

4.4 异步调用

同步 vs 异步

同步(sync)：

概念：发起一个任务之后，需要等待该任务完成后，才能继续执行后续任务。
表现：当前执行流会被『阻塞』。

异步(async)

概念：发起一个任务之后，不必等该任务完成，就可以继续执行其他任务
备注：虽然不必等待任务完成，但任务完成后，仍然可以通过特定方式获取结果
表现：当前执行流不会被『阻塞』。

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

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

4.4.1 使用ainvoke()

import asyncio
import time

from langchain_qwq import ChatQwen
from dotenv import load_dotenv
import os
load_dotenv(override=True)
qwen_model = ChatQwen(
    model="qwen-plus",
    api_key=os.getenv("QWEN_API_KEY"),
    base_url=os.getenv("QWEN_BASE_URL")
)
async def demo_async_invoke():
    print("===演示：ainvoke 的异步效果 ===")
    # 记录开始时间
    start_time = time.perf_counter()
    print("程序开始...")

    #1.创建任务 Task
    print(f">>> 发起异步大模型调用=== (ainvoke)...")
    async_task = asyncio.create_task(qwen_model.ainvoke("用一句话解释人工智能"))
    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())

4.4.2 使用astream()

import asyncio
import time

from langchain_core.utils.aiter import aclosing
from langchain_qwq import ChatQwen
from dotenv import load_dotenv
import os

from pygments.lexer import default

load_dotenv(override=True)
qwen_model = ChatQwen(
    model="qwen-plus",
    api_key=os.getenv("QWEN_API_KEY"),
    base_url=os.getenv("QWEN_BASE_URL")
)
async def demo_async_stream():
    # 演示异步调用的非阻塞特性
    print(f"=== 演示：astream 的异步（非阻塞）效果 ===")
    # 记录开始时间
    start_time = time.perf_counter()
    print(f"程序开始...")
    # 1.发起异步流式请求，注意：此时请求已发出，返回的是一个异步生成器
    print(f">>>发起异步流式调用(astream)...")
    astream_response = qwen_model.astream("请介绍什么是机器学习")

    # 2.在等待流式响应的同时，执行其他任务
    print(f">>>流式请求已发送，程序无需等待，继续执行其他异步任务...")
    for i in range(3):
        # 使用asyncio.sleep 而非 time.sleep
        # 允许时间循环在等待时去处理上面的astream_response 网络 IO
        await asyncio.sleep(1)
        print(f">>>正在执行第{i + 1} 个任务...(已耗时{time.perf_counter() - start_time:.2f}s)")
    # 3.开始处理流式结果
    print(f">>>模拟任务已完成，开始读取缓冲区中的流式结果...")
    end_time = time.perf_counter()
    print(">>> 流式输出: ", end = "",flush=True)
    async with aclosing(astream_response) as astream_response:
        async for chunk in astream_response:
            # LangChain的消息块通常通过.content 获取人内容
            content = chunk.content if hasattr(chunk,'content') else str(chunk)
            print(content,end="",flush=True)

    print(f"\n>>>流式输出结果\n")
    print(f"=== 总运行耗时: {end_time - start_time:.2f}s===")

async  def main():
    await demo_async_stream()
if __name__ == "__main__":
    asyncio.run(main())

4.4.3 使用abatch()

import asyncio
import time

from langchain_core.utils.aiter import aclosing
from langchain_qwq import ChatQwen
from dotenv import load_dotenv
import os

from pygments.lexer import default

load_dotenv(override=True)
qwen_model = ChatQwen(
    model="qwen-plus",
    api_key=os.getenv("QWEN_API_KEY"),
    base_url=os.getenv("QWEN_BASE_URL")
)
async def demo_async_stream():
    # 演示异步调用的非阻塞特性
    print(f"=== 演示：astream 的异步（非阻塞）效果 ===")
    # 记录开始时间
    start_time = time.perf_counter()
    print(f"程序开始...")
    # 1.发起异步流式请求，注意：此时请求已发出，返回的是一个异步生成器
    print(f">>>发起异步流式调用(astream)...")
    astream_response = qwen_model.astream("请介绍什么是机器学习")

    # 2.在等待流式响应的同时，执行其他任务
    print(f">>>流式请求已发送，程序无需等待，继续执行其他异步任务...")
    for i in range(3):
        # 使用asyncio.sleep 而非 time.sleep
        # 允许时间循环在等待时去处理上面的astream_response 网络 IO
        await asyncio.sleep(1)
        print(f">>>正在执行第{i + 1} 个任务...(已耗时{time.perf_counter() - start_time:.2f}s)")
    # 3.开始处理流式结果
    print(f">>>模拟任务已完成，开始读取缓冲区中的流式结果...")
    end_time = time.perf_counter()
    print(">>> 流式输出: ", end = "",flush=True)
    async with aclosing(astream_response) as astream_response:
        async for chunk in astream_response:
            # LangChain的消息块通常通过.content 获取人内容
            content = chunk.content if hasattr(chunk,'content') else str(chunk)
            print(content,end="",flush=True)

    print(f"\n>>>流式输出结果\n")
    print(f"=== 总运行耗时: {end_time - start_time:.2f}s===")

async  def main():
    await demo_async_stream()
if __name__ == "__main__":
    asyncio.run(main())

DeepSeek技术社区

欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。

更多推荐

Google能索引，AI却抓不到

很多技术团队有个错觉：只要Google能正常收录，内容就没问题。但现实是，你的页面在Google搜索结果里好好的，在ChatGPT、Perplexity、Claude的答案里却像从未存在过一样。

DeepSeek技术社区

初识智能体

在过去两年中，几乎所有人都体验过大语言模型（如 ChatGPT、Claude、DeepSeek）的强大能力：写周报、润色邮件、解答编程疑难。被动响应：你问一句，它答一句。如果你的 Prompt 没有写清楚约束条件，它无法主动追问或修正目标。知行脱节：大模型能用流畅的文字告诉你“如何部署一个高可用 K8s 集群”，但它无法替你登录服务器敲下命令、查看报错日志并自动回滚。信息孤岛：大模型的知识止步于预