介绍 LangChain 核心基础概念，包含模型配置初始化、系统提示词模板、LCEL 链式编排、智能体构建（create_agent）、工具定义与参数校验、持久化检查点、结构化输出及流式输出等特性。

框架选择：何时使用 LangChain 智能体

LangChain、LangGraph 与 Deep Agents 呈层级递进关系：

• Deep Agents（应用层）：提供规划、记忆、技能与文件管理等高级功能。
• LangGraph（编排层）：提供图结构、循环控制、状态管理与持久化能力。
• LangChain（基础层）：提供模型调用、工具定义、提示词模板与 RAG 组件。

LangChain 是该组件体系的基础，包含模型调用、提示词模板、结构化输出与工具调用等核心模块，适用于构建单一用途的 ReAct（Reasoning and Acting，即推理与行动）智能体或 RAG 文档问答应用。

依赖安装

LangChain 1.0 为长期支持（LTS）版本，要求 Python 3.10+ 环境。

# requirements.txt
langchain>=1.0,<2.0
langchain-core>=1.0,<2.0

# 根据模型提供商安装对应扩展包，例如：
langchain-openai

其中，langchain-core 为基础依赖，需显式安装。推荐使用独立版本控制的专用扩展包（如 langchain-openai）；若需安装未遵循语义化版本的 langchain-community 包，需锁定小版本号以降低接口变动风险。

使用 create_agent 创建智能体

create_agent() 用于构建智能体，统一处理智能体循环、工具执行与状态管理。该函数返回 CompiledStateGraph 对象，内置状态管理机制。

智能体配置选项说明：

参数	说明	示例
model	指定使用的 LLM	"anthropic:claude-sonnet-4-5" 或模型实例
tools	工具函数列表	[search, calculator]
system_prompt / systemPrompt	智能体系统指令	"You are a helpful assistant"
checkpointer	状态持久化后端	MemorySaver()
middleware	处理钩子（Hooks）	[HumanInTheLoopMiddleware]

from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def get_weather(location: str) -> str:
    """获取指定地点的当前天气。

    Args:
        location: 城市名称
    """
    return f"Weather in {location}: Sunny, 72F"

agent = create_agent(
    model="anthropic:claude-sonnet-4-5",
    tools=[get_weather],
    system_prompt="You are a helpful assistant."
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "What's the weather in Paris?"}]
})
print(result["messages"][-1].content)

模型配置

BaseModel

在通过 OpenAI 兼容协议调用模型外，各厂商模型常包含特定拓展特性。LangChain 提供了统一的 BaseModel 接口，将模型输入输出标准化为 BaseMessage 对象，以便在不同模型实现间进行切换。

# 读取 OPENAI_API_BASE 与 OPENAI_API_KEY 环境变量
llm = ChatOpenAI(
    model="qwen3.5-flash",
    temperature=0,
    streaming=True,
)
res = llm.invoke("hello")
print(res.content)

from langchain_qwq import ChatQwen

# 读取 DASHSCOPE_API_BASE（默认为阿里云国际站点，国内站点需单独配置）与 DASHSCOPE_API_KEY 环境变量
llm = ChatQwen(
    model="qwen3.5-flash",
    temperature=0,
    streaming=True,
)
res = llm.invoke("hello")
print(res.content)

init_chat_model 为 LangChain 提供的工厂函数，用于动态初始化指定提供商的聊天模型。调用此函数与直接实例化 ChatOpenAI 效果一致（指定 OpenAI 提供商时，底层返回 ChatOpenAI 实例）。

# 动态初始化实例
from langchain.chat_models import init_chat_model
llm = init_chat_model("qwen3.5-flash", model_provider="openai", temperature=0)

temperature 参数对比

• temperature=0（确定性输出模式）：响应逻辑确定且结果一致，适用于代码生成、实体提取与逻辑推理等任务。
• temperature=1（创造性输出模式）：响应随机性较高且思维发散，适用于创意写作与文本生成等场景。

自定义回调处理器 (BaseCallbackHandler) 与运行时配置

可在模型调用流程中插入回调以追踪执行状态，并通过 config 动态传入参数：

from langchain_core.callbacks import BaseCallbackHandler

class SimpleTrackingCallbackHandler(BaseCallbackHandler):
    def on_chat_model_start(self, serialized, messages, **kwargs):
        print("[Callback] 大模型调用开始...")

llm.invoke("你好", config={"callbacks": [SimpleTrackingCallbackHandler()]})

create_agent 模型配置

create_agent 支持通过格式为 "provider:model" 的字符串指定模型（例如 "openai:qwen3.5-flash"）。

系统会自动读取 OPENAI_API_BASE 与 OPENAI_API_KEY 环境变量，因此可直接传入 openai:qwen3.5-flash 标识符，无需显式实例化 ChatOpenAI 并配置参数。

若模型尚未注册 provider 前缀，可直接传入配置好的模型对象实例：

from langchain_qwq import ChatQwen

agent = create_agent(model=ChatQwen(model="qwen3.5-flash", temperature=0), tools=[...])

系统提示词

系统提示词定义智能体的身份角色、核心职责与行为边界，是控制智能体行为的基础方式。系统提示词通常需包含角色设定与工具使用规则，以规避模型幻觉或非预期工具调用。

def get_system_prompt():
    return (
        "你是一个专门为个人技术博客设计的对话式学习助手。\n\n"
        "### 核心职责：\n"
        "1. 基于博客内容回答技术问题。\n"
        "2. 使用工具搜索文章、获取统计数据。\n\n"
        "### 工具使用规则：\n"
        "- 在回答关于博客具体文章内容的问题前，必须先调用 search_blog 工具。禁止在未获得检索结果时自行推测回答。\n"
        "- 如果工具没有返回相关信息，明确说明没有找到，不要编造。\n"
    )

提示词模板

ChatPromptTemplate 用于管理多角色提示词结构，支持变量注入与角色设定。

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个翻译专家，请将以下文本翻译为{target_language}。"),
    ("user", "{text}")
])

局部变量绑定 (partial)

当提示词模板的部分变量需在初始化阶段固定（例如静态日期 current_date），而其余变量在运行时动态传入时，可通过 .partial() 预先绑定局部变量：

from langchain_core.prompts import PromptTemplate

prompt_template = PromptTemplate.from_template("今天是 {current_date}。话题: {topic}")
# 预先绑定局部参数
partial_prompt = prompt_template.partial(current_date="2026-05-27")
# 运行时传入剩余变量
chain = partial_prompt | model

聊天历史占位符 (MessagesPlaceholder)

在聊天提示词模板中，使用 MessagesPlaceholder("variable_name") 作为占位符，用于接收包含 HumanMessage 与 AIMessage 的对话历史列表，实现多轮对话上下文管理：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个客服。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}")
])

少样本提示词 (Few-Shot)

当零样本（Zero-Shot）提示无法满足格式或逻辑要求时，可通过少样本（Few-Shot）示例指导模型输出预期格式：

from langchain_core.prompts import FewShotChatMessagePromptTemplate

examples = [
    {"input": "2+2", "output": "4"},
    {"input": "2+3", "output": "5"},
]
# 定义单个样例的结构
example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

# 组装少样本模板
few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples,
)

# 嵌入到最终的提示词模板中
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个计算器。"),
    few_shot_prompt,
    ("human", "{input}"),
])

在 create_agent 中集成 ChatPromptTemplate

create_agent 的 system_prompt 参数支持 str 与 SystemMessage。若使用 ChatPromptTemplate 组织提示词，可通过以下方式进行集成：

1. 静态渲染：若模板变量在创建时即已固定，可先完成渲染再传入 create_agent。

   prompt = ChatPromptTemplate.from_messages([("system", "你是{role}。")])
   system_str = prompt.format_messages(role="专家")[0].content
   agent = create_agent(model="...", system_prompt=system_str)

2. 动态渲染：若模板变量依赖单次请求的运行时上下文（例如动态知识库），需使用 @dynamic_prompt 中间件进行拦截与动态生成。

   @dynamic_prompt
   def dynamic_system_prompt(request: ModelRequest) -> str:
       context = request.runtime.context
       return prompt.format_messages(role=context.get("role"))[0].content

LCEL 链

在无需复杂工具调用的单次或线性任务场景下，可使用 LCEL (LangChain Expression Language) 构建轻量化处理链（Chain）。

LCEL 通过管道操作符 | 连接组件（Prompt、Model、Parser 等）。常用组合结构为：提示词 | 模型 | 输出解析器。

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="qwen3.5-flash")
# StrOutputParser 用于将模型返回的 Message 对象转换为字符串
parser = StrOutputParser()

# 构建 LCEL 链
chain = prompt | llm | parser

# 执行调用
result = chain.invoke({
    "target_language": "法语",
    "text": "你好，世界！"
})
print(result)  # 直接输出字符串结果

Runnable 核心方法对比

LCEL 组件及组合生成的 Chain 均实现 Runnable 接口。核心方法说明如下：

方法	说明	适用场景
invoke	传入单个输入，同步阻塞等待并返回结果。	单次执行的简单脚本或同步任务。
ainvoke	传入单个输入，异步非阻塞返回结果。	Web 服务（如 FastAPI）及并发场景。
stream	传入单个输入，返回同步迭代器，逐块（Chunk）读取生成结果。	命令行控制台实时输出。
astream	传入单个输入，返回异步迭代器，使用 async for 遍历。	构建实时打字机效果的 Web API (SSE / WebSocket)。
batch	传入输入列表，利用线程池并发执行并返回结果列表。	批量数据同步处理。
abatch	传入输入列表，利用 asyncio 并发执行并返回结果列表。	异步批量数据处理。

LCEL 的主要特点：

1. 内置流式输出与异步调用能力（stream / ainvoke）。
2. 支持数据流可视化与声明式组装。
3. 支持插入自定义处理节点。

高级 LCEL 组合器 (Runnables)

除顺序组合操作符 | 外，LangChain 还提供了以下组合原语：

1. RunnableParallel（并行分叉）：
   将输入并发分发至多个分支执行，并将结果汇总为字典。适用于并行调用独立工具或模型的场景。

   chain = RunnableParallel({"summary": summary_chain, "translation": translate_chain})

2. RunnablePassthrough（数据透传）：
   传递原始输入或补充新增字段，常用于在 RAG 场景中保留原始用户提问（question）的同时注入检索上下文（context）。

   chain = {"context": retriever, "question": RunnablePassthrough()} | prompt | model

3. RunnableLambda（自定义逻辑）：
   将标准 Python 函数封装为可集成至 LCEL 链中的组件。

   def my_func(text: str) -> str:
       return text.upper()
   chain = prompt | model | RunnableLambda(my_func)

LCEL 对话历史 (RunnableWithMessageHistory)

在基础 LCEL 链中追加对话历史管理能力时，可使用 RunnableWithMessageHistory 进行封装：

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

store = {}
def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# chain 中需使用 MessagesPlaceholder("history") 接收注入的历史记录
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="question",
    history_messages_key="history",
)
chain_with_history.invoke({"question": "嗨！"}, config={"configurable": {"session_id": "u1"}})

获取调用结果

从返回字典中提取 messages 列表，并通过最后一条消息对象的 content 字段获取生成内容：

# 从结果字典中访问 messages
result = agent.invoke({"messages": [{"role": "user", "content": "Hello"}]})
print(result["messages"][-1].content)  # 获取最后一条消息的内容

批量调用 (Batch)

除了单次调用（invoke / ainvoke）外，create_agent 创建的智能体支持批量调用，用于并发处理多个独立对话任务：

• batch：同步批量调用，底层基于线程池并发执行。
• abatch：异步批量调用，底层基于 asyncio 并发执行。

# 同步批量调用
batch_results = agent.batch([
    {"messages": [{"role": "user", "content": "What's the weather in Paris?"}]},
    {"messages": [{"role": "user", "content": "What's the weather in Tokyo?"}]}
])
for res in batch_results:
    print(res["messages"][-1].content)

# 异步批量调用
import asyncio

async def main():
    abatch_results = await agent.abatch([
        {"messages": [{"role": "user", "content": "What's the weather in London?"}]},
        {"messages": [{"role": "user", "content": "What's the weather in Beijing?"}]}
    ])
    for res in abatch_results:
        print(res["messages"][-1].content)

asyncio.run(main())

递归限制 (recursion_limit)

可在调用配置中设置 recursion_limit 参数，明确限制智能体单次调用的最大迭代步数，避免出现无限循环：

# 未配置递归限制时，可能在异常情况下持续循环
result = agent.invoke({"messages": [("user", "Do research")]})

# 配置 recursion_limit 参数以限定最大步数
result = agent.invoke(
    {"messages": [("user", "Do research")]},
    config={"recursion_limit": 10},  # 达到 10 步后终止执行
)

定义工具

工具是智能体可调用的具体函数。可通过 @tool 装饰器定义。需要为工具编写具体清晰的文档字符串（docstring）及 Args 说明，以便智能体准确识别工具触发条件。

@tool
def search(query: str) -> str:
    """在网络上搜索有关某个话题的最新信息。

    当您需要最新数据或事实时使用此工具。

    Args:
        query: 搜索查询词（建议 2-10 个词）
    """
    return web_search(query)

基于 Pydantic 的工具参数校验 (args_schema)

除文档字符串外，可通过继承 Pydantic 的 BaseModel 定义参数结构，并传入 args_schema 进行强类型校验与约束：

from pydantic import BaseModel, Field

class GetWeatherArgs(BaseModel):
    location: str = Field(description="查询天气的城市名称")
    date: str = Field(default="today", description="查询的日期描述")

@tool(args_schema=GetWeatherArgs)
def get_weather(location: str, date: str) -> str:
    """获取指定地点的当前天气。"""
    return f"[{date}] {location} Weather: Sunny, 22C"

访问工具对象的 .name、.description、.args_schema 等属性，可查看导出的底层 StructuredTool 描述信息。

原生 Function Calling 流程

在未使用 create_agent 时，可以通过底层 bind_tools 接口直接绑定工具并手动处理调用流程：

# 1. 绑定工具
llm_with_tools = model.bind_tools([search])

# 2. 调用模型
ai_msg = llm_with_tools.invoke("搜索 LangChain 的新闻")

# 3. 解析模型返回的 tool_calls
if ai_msg.tool_calls:
    for tool_call in ai_msg.tool_calls:
        # 4. 执行对应的工具函数
        result = search.invoke(tool_call["args"])
        
        # 5. 将结果封装为 ToolMessage 组装回对话历史
        tool_msg = ToolMessage(content=result, tool_call_id=tool_call["id"], name=tool_call["name"])

持久化检查点 (Checkpointer)

通过传入 checkpointer 检查点组件，可在多次调用间维护对话状态。MemorySaver 提供了内存级别的状态保存功能。

检查点机制内部使用 thread_id 隔离会话。多用户会话场景下可使用 user_id:session_id 的组合键（例如 "user_456:session_888"）：

from langchain.agents import create_agent
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

agent = create_agent(
    model="openai:qwen3.5-flash",
    tools=[],
    checkpointer=checkpointer,
)
config = {"configurable": {"thread_id": "session-1"}}
agent.invoke({"messages": [{"role": "user", "content": "I'm Bob"}]}, config=config)
result = agent.invoke({"messages": [{"role": "user", "content": "What's my name?"}]}, config=config)
print(result["messages"][-1].content)

关于更多持久化后端（如 Redis/PostgreSQL 等），可参考官方文档。注意：部分第三方 Redis 检查点库要求 Redis 8.x+ 版本，如需在 Redis 7.x 环境运行，可继承 BaseCheckpointSaver 自定义实现 latest 检查点存储机制。

结构化输出

配置 response_format 参数可获取类型安全的结构化响应。

通过 ToolStrategy 生成 JSON 输出时，系统会通过强制指定 tool_choice=required 要求模型返回结构化数据。若使用 qwen3.5-flash 等带有默认思考过程的模型，需通过 extra_body 关闭思考模式，防止格式解析异常：

from langchain.agents import create_agent
from pydantic import BaseModel, Field

class ContactInfo(BaseModel):
    name: str
    email: str
    phone: str = Field(description="带有区号的电话号码")

model = init_chat_model(
    "openai:qwen3.5-flash",
    extra_body={"enable_thinking": False},
)

agent = create_agent(
    model=model,
    tools=[search],
    response_format=ToolStrategy(ContactInfo),
)
result = agent.invoke({"messages": [{"role": "user", "content": "Find contact for John"}]})
print(result["structured_response"])  # ContactInfo(name='John', ...)

其他结构化解析器 (Output Parsers)

在普通的 LCEL 链中提取结构化数据，主要包含以下方式：

1. with_structured_output 结构化提取

直接在模型对象上调用该方法，返回自动提取并解析为 Pydantic 对象的 Runnable 组件：

structured_llm = llm.with_structured_output(Person)
chain = prompt | structured_llm
result = chain.invoke("张三今年32岁...")  # 返回 Person 实例对象

2. Pydantic 字段校验器 (@field_validator)

在 Pydantic 模型类中使用 @field_validator 对字段内容进行自定义业务校验，校验失败时抛出 ValueError：

from pydantic import BaseModel, field_validator

class Person(BaseModel):
    age: int

    @field_validator("age")
    def validate_age(cls, v):
        if v < 0 or v > 120:
            raise ValueError("Invalid age range")
        return v

3. 标准结构化解析器

• PydanticOutputParser：结合 parser.get_format_instructions() 注入 Prompt 并解析为 Pydantic 实例。
• JsonOutputParser：直接将输出解析为标准 Python 字典对象（dict）。

4. 自定义解析器 (BaseOutputParser)

继承 BaseOutputParser[T] 并实现 parse(self, text: str) -> T 方法，定制解析逻辑。

流式输出

流式输出用于在响应生成过程中逐步向前端传输数据，提供增量式反馈。

调用 agent.stream() 或 agent.astream()，并通过 stream_mode 参数配置流式类型：

• "updates"：每个处理步骤（模型/工具）执行完成后的状态更新。
• "messages"：LLM 输出的逐 Token 数据流。
• "custom"：工具内部通过 get_stream_writer() 发送的自定义事件。

支持多模式组合：stream_mode=["updates", "messages", "custom"]

智能体步骤更新 (updates)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "北京天气怎么样？"}]},
    stream_mode="updates",
    version="v2",
):
    if chunk["type"] == "updates":
        for step, data in chunk["data"].items():
            print(f"[步骤: {step}] {data['messages'][-1].content_blocks}")

逐 Token 流式输出 (messages)

from langchain.messages import AIMessageChunk

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "上海天气如何？"}]},
    stream_mode="messages",
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]
        if isinstance(token, AIMessageChunk) and token.text:
            print(token.text, end="", flush=True)

自定义进度流 (custom)

在工具内部使用 get_stream_writer() 发送进度信息：

from langgraph.config import get_stream_writer

@tool
def get_weather(city: str) -> str:
    """获取天气"""
    writer = get_stream_writer()
    writer(f"正在查询 {city} 的天气...")
    return f"Weather in {city}: Sunny"

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "深圳天气怎么样？"}]},
    stream_mode="custom",
    version="v2",
):
    if chunk["type"] == "custom":
        print(f"[进度] {chunk['data']}")

说明：使用 get_stream_writer() 后，工具仅能在 LangGraph 执行上下文中调用。

stream 与 astream 调用对比

stream() 与 astream() 差异说明如下：

• stream()：同步方法，使用标准 for 循环遍历，适用于命令行脚本或测试环境。
• astream()：异步方法，使用 async for 遍历，适用于异步 Web 框架（如 FastAPI）。

在 Web 服务中推荐使用 astream()，避免同步调用阻塞事件循环。

import asyncio
from langchain.agents import create_agent
from langchain.messages import AIMessageChunk

agent = create_agent(model="openai:qwen3.5-flash", tools=[get_weather])

async def main():
    async for chunk in agent.astream(
        {"messages": [{"role": "user", "content": "北京天气怎么样？"}]},
        stream_mode="messages",
        version="v2",
    ):
        if chunk["type"] == "messages":
            token, metadata = chunk["data"]
            if isinstance(token, AIMessageChunk) and token.text:
                print(token.text, end="", flush=True)

asyncio.run(main())

FastAPI 结合 StreamingResponse 示例：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/chat")
async def chat(query: str):
    async def generate():
        async for chunk in agent.astream(
            {"messages": [{"role": "user", "content": query}]},
            stream_mode="messages",
            version="v2",
        ):
            if chunk["type"] == "messages":
                token, _ = chunk["data"]
                if isinstance(token, AIMessageChunk) and token.text:
                    yield token.text
    return StreamingResponse(generate(), media_type="text/plain")

高级特性 (Advanced Features)

针对生产环境开发，LangChain 提供了以下底层增强特性：

1. 多模态图像输入 (Multimodal)

多模态模型支持处理图像等非文本输入。通过将图像做 Base64 编码并组装至 HumanMessage 的 content 列表中，可调用视觉模型（例如 qwen3-vl-flash）进行图像识别：

from langchain_core.messages import HumanMessage

message = HumanMessage(
    content=[
        {"type": "text", "text": "请识别此图片中的内容。"},
        {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64_string}"}}
    ]
)
response = vlm_model.invoke([message])

2. 速率限制器 (InMemoryRateLimiter)

为配合 API 供应商的调用频率限制（Rate Limit），可将 InMemoryRateLimiter 传入模型对象，平滑控制请求频率，规避 HTTP 429 错误：

from langchain_core.rate_limiters import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(requests_per_second=0.5, max_bucket_size=2)
llm = ChatOpenAI(model="qwen3.5-flash", rate_limiter=rate_limiter)

3. Token 使用度量与追踪 (Token Tracking)

使用以下方式追踪模型调用的 Token 消耗：

• 方式 A（回调管理器）：使用 get_usage_metadata_callback() 统计 Prompt、Completion 及 Total Token 消耗。
• 方式 B（响应元数据）：读取 Response 对象的 .usage_metadata 属性。

4. 模型能力反射

可通过底层字段在运行时校验和查询模型对象支持的参数及工具调用特性。