LangChain 是目前大模型应用开发领域最主流、最成熟的开源框架之一，广泛应用于私有化部署、智能体开发、工具调用、流式交互等各类场景。本文基于 LangChain 最新稳定版接口，适配本地私有化 GGUF 模型部署环境，拆解框架四大核心基础能力：标准化消息机制、链路工具调用链路、多场景流式传输、规范化结构化输出功能，这四个功能是学习后续框架的基础部分，也是必须要熟练掌握的重点。
原生大模型接口存在纯文本传输无角色区分、多轮对话无标准化上下文、工具调用链路不链路、输出格式自由混乱、长文本响应延迟过高，难以直接落地生产项目。而 LangChain 通过标准化封装，解决上述问题，屏蔽了不同大模型、不同部署方式、不同接口协议的差异化适配成本，让开发者可以聚焦业务逻辑开发。
本文基于 LangChain 稳定版接口，拆解 LangChain 四大基础能力：标准化消息机制、链路工具调用链路、多场景流式传输、规范化结构化输出。这些功能是框架的基石，也是大模型应用开发、智能体迭代、生产环境落地的必备核心技能，所有高阶框架能力均基于此拓展而来。
LangChain 消息机制
消息（Message）是 LangChain 框架的底层核心单元，大模型对话交互、多轮上下文记忆、工具数据传输、智能体流转等所有高级能力，本质都是各类消息对象的拼接、传递与更新。
传统原生大模型调用仅支持纯文本字符串传输，无法区分对话角色、无法携带运维元数据、无法适配多厂商模型接口差异、工具交互无标准化规范。而 LangChain 标准化消息体系解决了以上问题，实现多模型无感切换、对话状态持久化、工具数据链路传输，开发者无需针对不同模型单独适配接口。
LangChain 每一个标准消息对象均由角色、内容、元数据三要素组成，共同支撑完整的商业化对话交互能力：
- 角色（role）：核心标识字段，用于区分消息发送主体，严格区分系统、用户、AI、工具四类角色，模型会根据角色优先级解析对话逻辑，角色错乱会直接导致回答异常、工具调用失效。
- 内容（content）：消息的核心载荷数据，不仅支持普通文本，还兼容图片、音频、多模态混合数组、空内容等格式，适配纯文本问答、多模态识别等各类场景。
- 元数据（metadata）：可扩展可选字段，用于存储运维与监控数据，包含 Token 消耗统计、唯一消息ID、模型指纹、回答结束原因、缓存状态、请求耗时等，适配生产环境日志排查、性能监控、接口溯源需求。
LangChain 规范了四类核心消息类型，覆盖所有对话与工具交互场景：
- System message (系统消息)：告诉模型如何表现并为交互提供上下文
- Human message (人类消息)：代表用户输入以及与模型的交互
- AI message (AI 消息)：模型生成的响应，包括文本内容、工具调用和元数据
- Tool message (工具消息)：代表工具调用的输出
基本消息
LangChain 同时支持强类型对象写法和原生字典写法，两种写法功能完全一致，仅适配不同开发阶段与场景，开发者可按需选择：
- 强类型对象（生产环境首选）：基于 LangChain 内置消息类实例化，自带语法校验、代码提示、属性补全，规范度高、可读性强，可有效规避角色错乱、参数错误等问题，适合正式项目开发。
- 原生字典写法（调试原型首选）：完全兼容原生 OpenAI 接口格式，无需导入各类消息类，代码简洁、上手快速，适合快速验证功能、原型调试、临时测试场景。
强类型消息对象调用
该示例通过标准消息类构建对话上下文，初始化本地兼容模型，完成基础问答交互，代码结构规范，适合生产复用。
import json
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
if __name__ == "__main__":
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.7,
max_tokens=512,
)
# 消息
system_msg = SystemMessage(content="你是一个问答助手，你可以回答用户的问题")
human_msg = HumanMessage(content="你好")
ai_msg = AIMessage(content="你好")
# 执行
messages = [system_msg, human_msg]
response = llm.invoke(messages)
# AIMessage转字典 输出美化JSON
resp_dict = response.model_dump()
json_str = json.dumps(resp_dict, ensure_ascii=False, indent=2)
print(json_str)
运行后可以输出如下所示的 JSON 格式，其中就包含了完整的消息字段。
CMD> python main.py
{
"content": "你好！有什么可以帮到你的吗？",
"additional_kwargs": { "refusal": null },
"response_metadata": {
"token_usage": {
"completion_tokens": 10,
"prompt_tokens": 23,
"total_tokens": 33,
"completion_tokens_details": null,
"prompt_tokens_details": {
"audio_tokens": null,
"cache_write_tokens": null,
"cached_tokens": 22,
"image_tokens": null,
"text_tokens": null
}
},
"model_provider": "openai",
"model_name": "qwen2.5-1.5b-instruct-q4_k_m.gguf",
"system_fingerprint": "b10453-3cb7ffb1a",
"id": "chatcmpl-LlfKVJBDIQSRc69pbru5rMR3rZ5DYaH1",
"finish_reason": "stop",
"logprobs": null
},
"type": "ai",
"name": null,
"id": "lc_run--01a03c1e-7e88-7ea0-8a32-9e5fd9d9c1e8-0",
"tool_calls": [],
"invalid_tool_calls": [],
"usage_metadata": {
"input_tokens": 23,
"output_tokens": 10,
"total_tokens": 33,
"input_token_details": { "cache_read": 22 },
"output_token_details": {}
}
}
原生字典消息调用
复用原生 OpenAI 字典格式，无需导入消息实体类，简化代码结构，适合快速调试、功能验证场景。
import json
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
if __name__ == "__main__":
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.7,
max_tokens=512,
)
# 消息
messages = [
{"role": "system", "content": "你是一个问答助手，你可以回答用户的问题"},
{"role": "user", "content": "你好"},
{"role": "assistant", "content": "你好"}
]
# 执行
response = llm.invoke(messages)
# AIMessage转字典 输出美化JSON
resp_dict = response.model_dump()
json_str = json.dumps(resp_dict, ensure_ascii=False, indent=2)
print(json_str)
运行后可以输出如下所示的 JSON 格式，其中就包含了完整的消息字段。
CMD> python main.py
{ ... 同上 ... }
工具消息
大模型本身存在知识时效性滞后、无法操作外部资源、无法执行逻辑代码、无法获取实时数据等缺陷，而工具调用是智能体开发的核心能力，可让大模型突破自身能力限制，主动调用自定义函数、第三方接口、数据库、文件系统等外部资源，完成实时数据查询、逻辑计算、业务处理等复杂操作。
模型触发工具调用
本案例自定义天气查询工具，实现模型自主识别提问意图、自动触发工具调用，可打印完整工具调用参数，适配工具调试场景。
当模型进行工具调用时，这些调用会被包含在 AIMessage 中，其他结构化数据（如推理或引用）也可以出现在消息内容中。
import json
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
def get_weather(location: str) -> str:
"""获取地区天气"""
return f"在{location}的天气是晴朗的"
if __name__ == "__main__":
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.7,
max_tokens=512,
)
model_with_tools = llm.bind_tools([get_weather])
response = model_with_tools.invoke("济南天气怎么样？")
for tool_call in response.tool_calls:
print(f"Tool: {tool_call['name']}")
print(f"Args: {tool_call['args']}")
print(f"ID: {tool_call['id']}")
resp_dict = response.model_dump()
json_str = json.dumps(resp_dict, ensure_ascii=False, indent=2)
print(json_str)
运行后可以输出如下所示的 JSON 格式，其中就包含了完整的消息字段。
CMD> python main.py
Tool: get_weather
Args: {'location': '济南'}
ID: iCibl1XVYuLOu8nh7jXtjuyydn0zyflM
{ ... 完整的 JSON 输出 ... }
ToolMessage 链路实现
工具调用能否链路的唯一标准是 tool_call_id 匹配。ToolMessage 的 tool_call_id 必须与模型返回的工具调用 ID 完全一致，否则模型无法关联工具调用指令与执行结果，直接导致工具调用失效、对话链路失败、无最终答案输出。
本案例完整模拟工具调用全流程，手动拼接消息链路，实现从工具调用、结果封装、二次推理到最终输出的完整链路。
import json
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, ToolMessage
def get_weather(location: str) -> str:
"""获取地区天气"""
return f"在{location}的天气是晴朗的"
if __name__ == "__main__":
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.7,
max_tokens=512,
)
ai_message = AIMessage(
content=[],
tool_calls=[{
"name": "get_weather",
"args": {"location": "济南"},
"id": "call_1"
}]
)
weather_result = "晴朗,25C"
tool_message = ToolMessage(
content=weather_result,
tool_call_id="call_1"
)
messages = [
SystemMessage(content="你是一个天气助手"),
HumanMessage(content="济南的天气怎么样？"),
ai_message,
tool_message
]
response = llm.invoke(messages)
resp_json = json.dumps(response.model_dump(), ensure_ascii=False, indent=2)
print(resp_json)
运行后可以输出如下所示的 JSON 格式，其中就包含了完整的消息字段。
CMD> python main.py
{
"content": "济南今天的天气是晴朗，气温大约在25℃左右。",
... 其他字段 ...
}
LangChain 流式传输
传统阻塞式 invoke 调用需要等待模型生成全部内容完成后才会统一返回结果，大文本、长推理场景下延迟极高，用户交互体验极差。而流式传输（Streaming）支持模型逐 Token 增量输出数据，边生成、边返回、边渲染，大幅降低首屏响应时间，是 AI 对话页面、智能体可视化、实时问答系统的必备能力。
LangChain 流式传输的核心载体为 AIMessageChunk 分片对象，所有增量分片数据可自动合并、拼接为完整的 AIMessage，兼顾实时输出与结果完整性，支持文本、工具调用、模型思考过程多维度流式输出。
基础 LLM 文本流式输出
最基础、最常用的流式能力，适用于普通文本问答场景，逐字输出模型回答，无需等待完整生成，适配绝大多数对话页面展示需求。
import json
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
if __name__ == "__main__":
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.7,
max_tokens=512,
)
chunks = []
for chunk in llm.stream("你好呀？"):
chunks.append(chunk)
resp_dict = chunk.model_dump()
json_str = json.dumps(resp_dict, ensure_ascii=False, indent=2)
print(json_str)
智能体步骤流式监控
专门用于智能体复杂执行链路监控，可实时捕获工具调用触发、工具参数入参、工具执行结果、模型最终生成每一步状态，适合后端日志打印、前端进度条渲染、智能体流程可视化场景。
要流式传输智能体进度，请在调用 stream 或 astream 方法时设置 stream_mode="updates"。这会在每个智能体步骤后发射一个事件。
通过 config 传递 thread_id，以便保存对话检查点，并在后续轮次中可以恢复同一历史记录。
import json
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, ToolMessage
from langchain.agents import create_agent
from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.7,
max_tokens=512,
)
def get_weather(location: str) -> str:
"""获取地区天气"""
return f"在{location}的天气是晴朗的"
if __name__ == "__main__":
agent = create_agent(
model=llm,
tools=[get_weather],
checkpointer=InMemorySaver()
)
config = {"configurable": {"thread_id": str(uuid7())}}
stream = agent.stream_events(
{"messages": [{"role": "user", "content": "查询在济南的天气"}]},
config=config,
version="v3",
)
for kind, item in stream.interleave("messages", "tool_calls"):
if kind == "messages":
for token in item.text:
print(token, end="", flush=True)
elif kind == "tool_calls":
print(f"\nTool call: {item.tool_name}({item.input})")
for delta in item.output_deltas:
print(delta, end="", flush=True)
print(f"\nTool result: {item.output}")
final_state = stream.output["messages"][-1].content
print(final_state)
LLM Token 精细化流式输出
精细化流式模式，可精准拆分文本回答分片、工具调用分片、不同智能体节点输出，支持自定义分片解析逻辑，适合需要精细区分输出来源、定制化流式渲染的高阶场景。
要流式传输 LLM 生成的 Token，请使用 stream_mode="messages"。
import json
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.prebuilt import create_react_agent
from langchain_core.utils.uuid import uuid7
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.7,
max_tokens=512,
)
def get_weather(location: str) -> str:
"""获取地区天气"""
return f"在{location}的天气是晴朗的"
if __name__ == "__main__":
agent = create_react_agent(
model=llm,
tools=[get_weather],
)
config = {"configurable": {"thread_id": str(uuid7())}}
for token, metadata in agent.stream(
{"messages": [HumanMessage(content="济南天气如何?")]},
config=config,
stream_mode="messages"
):
print(f"node: {metadata['langgraph_node']}")
print(f"content_blocks: {token.content_blocks}")
print(f"text delta: {token.content}")
print("-" * 40)
模型推理思考过程流式输出
针对支持深度推理的大模型，支持单独流式输出模型思考过程，实现"推理过程"与"最终答案"分离展示，适配 AI 推理可视化、教学演示、智能体思维链路展示等场景。
某些模型在生成最终答案之前会进行内部推理。你可以通过过滤标准内容块中 type 为 "reasoning" 的内容，在生成思考/推理 Token 时对其进行流式传输。
import warnings
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain.agents import create_agent
from langchain_core.utils.uuid import uuid7
warnings.filterwarnings("ignore")
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.7,
max_tokens=512,
streaming=True,
timeout=None,
stop=None,
extra_body={
"thinking": {"type": "enabled", "budget_tokens": 5000}
}
)
def get_weather(location: str) -> str:
"""获取地区天气"""
return f"在{location}的天气是晴朗的"
if __name__ == "__main__":
agent = create_agent(
model=llm,
tools=[get_weather],
)
config = {"configurable": {"thread_id": str(uuid7())}}
stream = agent.stream_events(
{"messages": [HumanMessage(content="济南的天气如何?")]},
config=config,
version="v3",
)
for message in stream.messages:
print("[思考]: ", end="")
for token in message.reasoning:
print(token, end="", flush=True)
print("\n[回答]: ", end="")
for token in message.text:
print(token, end="", flush=True)
print("\n" + "-" * 50)
LangChain 结构化输出
大模型原生输出为自由格式文本，存在格式混乱、解析困难、容错率低的问题，业务开发中需要编写大量正则、字符串切割、异常兼容代码，维护成本极高。而 LangChain 结构化输出能力，可强制模型严格按照开发者预设的格式、字段、类型、约束返回数据，直接输出标准化结构化对象，无需手动解析，完美适配接口开发、数据入库、表单信息提取、内容分类、数据统计等生产场景。
LangChain 提供四种成熟的结构化输出方案，覆盖轻量化调试、生产校验、动态适配、跨语言对接全场景。
Pydantic Model
依托 Pydantic 强类型校验能力，可定义字段类型、描述、取值范围、默认值，模型输出后自动校验格式合法性，格式错误直接抛出异常，从源头保证数据可靠性，是生产环境标准方案。
第一种方法，使用 with_structured_output 绑定结构化输出并调用模型。
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.7,
max_tokens=512
)
class ContactInfo(BaseModel):
"""一个人的联系信息。"""
name: str = Field(description="该人的姓名")
email: str = Field(description="该人的电子邮件地址")
phone: str = Field(description="该人的电话号码")
if __name__ == "__main__":
structured_llm = llm.with_structured_output(ContactInfo)
result = structured_llm.invoke("请提供我的联系信息：王瑞，邮箱wang@test.com，电话13800000000")
print("姓名:", result.name)
print("邮箱:", result.email)
print("电话:", result.phone)
dataclass Model
Python 原生数据类，语法简洁轻量化，无需额外配置，适合简单场景快速封装数据，缺点是无运行时校验，仅依靠注释约束模型输出。
from dataclasses import dataclass
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.0,
max_tokens=512
)
@dataclass
class ContactInfo:
"""一个人的联系信息：包含姓名、邮箱、电话号码"""
name: str
email: str
phone: str
tools = []
user_prompt = "请提供我的联系信息：王瑞，邮箱wang@test.com，电话13800000000"
if __name__ == "__main__":
agent = create_agent(
model=llm, tools=tools, response_format=ContactInfo
)
result = agent.invoke({"messages": [{"role": "user", "content": user_prompt}]})
structured = result["structured_response"]
print(f"姓名={structured.name}, 邮箱={structured.email}, 电话={structured.phone}")
TypedDict Model
基于字典的轻量化结构化方案，无第三方依赖，输出原生字典格式，可直接用于接口返回，适合快速开发、轻量化调试场景。
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.0,
max_tokens=512
)
class ContactInfo(TypedDict):
"""一个人的联系信息"""
name: str
email: str
phone: str
tools = []
user_prompt = "请提供我的联系信息：王瑞，邮箱wang@test.com，电话13800000000"
if __name__ == "__main__":
agent = create_agent(
model=llm, tools=tools, response_format=ContactInfo
)
result = agent.invoke({"messages": [{"role": "user", "content": user_prompt}]})
structured = result["structured_response"]
print(f"姓名={structured['name']}, 邮箱={structured['email']}, 电话={structured['phone']}")
JSON Schema Model
手动定义标准 JSON 结构，通用性最强，支持动态生成 Schema、跨语言适配、自定义复杂嵌套结构，适合复杂动态格式场景。
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.0,
max_tokens=512
)
contact_info_schema = {
"type": "object",
"description": "一个人的联系信息。",
"properties": {
"name": {"type": "string", "description": "该人的姓名"},
"email": {"type": "string", "description": "该人的电子邮件地址"},
"phone": {"type": "string", "description": "该人的电话号码"}
},
"required": ["name", "email", "phone"]
}
tools = []
user_prompt = "请提供我的联系信息：王瑞，邮箱wang@test.com，电话13800000000"
if __name__ == "__main__":
agent = create_agent(
model=llm, tools=tools, response_format=contact_info_schema
)
result = agent.invoke({"messages": [{"role": "user", "content": user_prompt}]})
structured = result["structured_response"]
print(f"姓名={structured['name']}, 邮箱={structured['email']}, 电话={structured['phone']}")
ToolStrategy
部分轻量化本地模型、老旧开源模型不支持原生结构化输出能力，直接使用 with_structured_output 会出现格式错乱、返回文本不规范、报错等问题。
针对该兼容问题，LangChain 提供 ToolStrategy 兜底方案，将结构化 Schema 伪装成自定义工具，强制模型触发工具调用，通过工具返回结果实现标准化结构化输出，可兼容所有支持工具调用的大模型，适配本地私有化轻量化模型部署场景。
from pydantic import BaseModel, Field
from typing import Literal
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
llm = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="http://127.0.0.1:11433/v1",
api_key="dummy",
temperature=0.0,
max_tokens=512
)
class ProductReview(BaseModel):
"""对产品评论的分析。"""
rating: int | None = Field(description="产品的评分", ge=1, le=5)
sentiment: Literal["积极的", "负面的"] = Field(description="评论的情感倾向")
key_points: list[str] = Field(description="评论的要点。小写，每条 1-3 个词。")
tools = []
if __name__ == "__main__":
agent = create_agent(
model=llm, tools=tools,
response_format=ToolStrategy(ProductReview)
)
result = agent.invoke({
"messages": [{"role": "user", "content": "分析这篇评论：\"很棒的产品：5颗星。快速发货，但价格昂贵。"}]
})
print(result["structured_response"])
# 运行输出：rating=5 sentiment='积极的' key_points=['很棒的产品', '快速发货', '价格昂贵']
评论0