LangChain应用开发完全教程

一、概述与核心架构

1.1 什么是LangChain

LangChain是一个用于构建大语言模型（LLM）应用的开源框架，由Harrison Chase于2022年10月首次发布。它提供了一套标准化的接口、组件和工具链，帮助开发者将LLM与外部数据源、工具和工作流高效集成，从而构建强大的AI应用。LangChain的核心理念是"链式组合"——将不同的组件像链条一样连接起来，形成复杂的处理流程。

与直接调用LLM API不同，LangChain抽象了底层模型的差异，提供了统一的编程接口。无论你使用OpenAI、Anthropic、Google Gemini还是本地部署的开源模型，都可以用相同的代码结构进行切换。这种设计使得开发者能够专注于业务逻辑，而非底层模型的适配工作。

1.2 核心架构

LangChain的架构由以下几个核心层次组成：

• langchain-core：基础抽象层，定义了所有核心接口（Runnable接口、消息类型等）
• langchain-community：社区贡献的第三方集成组件（各种LLM、工具、向量数据库等）
• langchain：高级链（Chain）和Agent抽象，面向应用层的编排逻辑
• LangGraph：基于图的有状态编排框架，用于构建复杂的多步骤Agent
• LangSmith：调试、追踪、评估和监控平台

这种分层设计使得框架既保持了灵活性，又具备了良好的可扩展性。开发者可以根据需要选择使用底层组件或高层抽象。

1.3 安装与环境配置

# 安装LangChain核心包
pip install langchain langchain-core langchain-community

# 安装OpenAI集成
pip install langchain-openai

# 安装Anthropic集成
pip install langchain-anthropic

# 安装向量数据库（示例：FAISS）
pip install faiss-cpu

# 安装文档加载器
pip install pypdf unstructured

# 安装LangSmith SDK（可选）
pip install langsmith

配置环境变量：

import os

# OpenAI API配置
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

# LangSmith配置（可选，用于调试和追踪）
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-project"

注意：在生产环境中，应使用环境变量管理工具（如dotenv）或密钥管理服务来保护API密钥，切勿将其硬编码在源代码中。

---

二、模型接口（Chat Models / LLMs / Embeddings）

2.1 Chat Models

Chat Models是LangChain中最推荐使用的模型接口，专为对话场景设计。它们接受一系列消息作为输入，并返回一条消息作为输出。

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

# 初始化Chat Model
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.7,
    max_tokens=2000
)

# 基本调用
messages = [
    SystemMessage(content="你是一个专业的Python开发者助手。"),
    HumanMessage(content="请解释Python中的装饰器是什么？")
]

response = llm.invoke(messages)
print(response.content)

LangChain支持多种消息类型：

• SystemMessage：系统提示，设定AI的行为和角色
• HumanMessage：用户输入
• AIMessage：AI的回复
• ToolMessage：工具调用的结果
• FunctionMessage：函数调用的结果（旧版兼容）

2.2 LLMs（纯文本模型）

LLMs是传统的文本补全接口，接受纯文本字符串输入并返回字符串输出。虽然在新版本中推荐使用Chat Models，但理解LLMs接口有助于处理某些遗留系统。

from langchain_openai import OpenAI

# 初始化纯文本LLM
llm = OpenAI(
    model="gpt-3.5-turbo-instruct",
    temperature=0.7,
    max_tokens=1000
)

# 纯文本调用
response = llm.invoke("请用三句话解释量子计算的基本原理：")
print(response)

2.3 Embeddings（嵌入模型）

Embeddings模型将文本转换为高维向量表示，是RAG（检索增强生成）系统的核心组件。LangChain提供了统一的Embeddings接口。

from langchain_openai import OpenAIEmbeddings

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=1536
)

# 单个文本嵌入
vector = embeddings.embed_query("LangChain是一个强大的LLM应用框架")
print(f"向量维度: {len(vector)}")

# 批量文本嵌入
texts = [
    "LangChain用于构建LLM应用",
    "LlamaIndex专注于数据索引和检索",
    "向量数据库存储嵌入向量"
]
vectors = embeddings.embed_documents(texts)
print(f"生成了 {len(vectors)} 个向量")

2.4 模型切换与适配

LangChain的一大优势是模型无关性。切换模型只需更换初始化代码，业务逻辑无需修改：

# OpenAI
from langchain_openai import ChatOpenAI
llm_openai = ChatOpenAI(model="gpt-4o")

# Anthropic
from langchain_anthropic import ChatAnthropic
llm_claude = ChatAnthropic(model="claude-3-5-sonnet-20241022")

# 本地Ollama
from langchain_ollama import ChatOllama
llm_local = ChatOllama(model="llama3")

# 所有模型都使用相同的invoke接口
response = llm_openai.invoke("Hello!")
# 只需替换变量即可切换模型

---

三、Prompt模板与Few-shot Prompt

3.1 基本Prompt模板

Prompt模板是LangChain中构建动态提示的核心工具。它们允许你创建可复用的提示结构，通过变量填充来生成最终的提示文本。

from langchain_core.prompts import ChatPromptTemplate

# 创建聊天提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位{role}专家，请用{language}回答问题。"),
    ("human", "{question}")
])

# 格式化提示
messages = prompt.invoke({
    "role": "机器学习",
    "language": "中文",
    "question": "什么是过拟合？如何避免？"
})

# 与模型组合使用
llm = ChatOpenAI(model="gpt-4o")
chain = prompt | llm
response = chain.invoke({
    "role": "机器学习",
    "language": "中文",
    "question": "什么是过拟合？如何避免？"
})
print(response.content)

3.2 Few-shot Prompt模板

Few-shot Prompt通过提供少量示例来引导模型的输出格式和风格，特别适用于需要结构化输出的场景。

from langchain_core.prompts import FewShotChatMessagePromptTemplate

# 定义示例
examples = [
    {
        "input": "产品很好用，物流也快，非常满意！",
        "output": "正面"
    },
    {
        "input": "质量太差了，用了一天就坏了，后悔购买。",
        "output": "负面"
    },
    {
        "input": "一般般吧，没什么特别的，价格偏贵。",
        "output": "中性"
    }
]

# 创建示例模板
example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}")
])

# 创建Few-shot模板
few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples,
)

# 完整提示
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个情感分析专家。请对以下文本进行情感分类，只输出：正面、负面或中性。"),
    few_shot_prompt,
    ("human", "{input}")
])

# 使用
chain = final_prompt | llm
response = chain.invoke({"input": "这个产品性价比很高，推荐购买！"})
print(response.content)  # 输出: 正面

3.3 结构化输出（Structured Output）

LangChain支持让模型返回结构化的JSON输出，这对于需要程序化处理模型结果的场景非常重要。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# 定义输出结构
class ProductReview(BaseModel):
    sentiment: str = Field(description="情感分类：正面/负面/中性")
    confidence: float = Field(description="置信度，0-1之间")
    key_points: list[str] = Field(description="关键观点列表")
    summary: str = Field(description="一句话总结")

llm = ChatOpenAI(model="gpt-4o")
structured_llm = llm.with_structured_output(ProductReview)

prompt = ChatPromptTemplate.from_messages([
    ("system", "请分析以下产品评论，返回结构化的分析结果。"),
    ("human", "{review}")
])

chain = prompt | structured_llm
result = chain.invoke({
    "review": "这款手机拍照效果很棒，电池续航也不错，就是价格有点贵。系统偶尔会卡顿，但整体体验还是很好的。"
})

print(f"情感: {result.sentiment}")
print(f"置信度: {result.confidence}")
print(f"关键观点: {result.key_points}")
print(f"总结: {result.summary}")

---

四、链（Chain）设计模式

4.1 简单链

链是LangChain的核心概念，它将多个组件串联起来形成一个完整的处理流程。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 简单链：提示 -> 模型 -> 输出解析
prompt = ChatPromptTemplate.from_template(
    "请用一段话解释{concept}的概念，要求通俗易懂，适合初学者理解："
)

llm = ChatOpenAI(model="gpt-4o")
output_parser = StrOutputParser()

# 使用管道操作符组合链
chain = prompt | llm | output_parser

# 调用链
result = chain.invoke({"concept": "微服务架构"})
print(result)

4.2 顺序链（Sequential Chain）

顺序链将多个处理步骤按顺序执行，前一步的输出作为后一步的输入。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o")

# 步骤1：生成文章大纲
outline_prompt = ChatPromptTemplate.from_template(
    "请为以下主题生成一个详细的文章大纲（包含3-5个主要部分）：\n\n主题：{topic}"
)

# 步骤2：根据大纲撰写文章
article_prompt = ChatPromptTemplate.from_template(
    "请根据以下大纲撰写一篇详细的技术文章：\n\n大纲：{outline}\n\n要求：内容详实，包含代码示例，字数不少于500字。"
)

# 构建顺序链
outline_chain = outline_prompt | llm | StrOutputParser()
article_chain = article_prompt | llm | StrOutputParser()

# 方法1：手动串联
outline = outline_chain.invoke({"topic": "Python异步编程"})
article = article_chain.invoke({"outline": outline})
print(article)

4.3 分支链（Router Chain）

分支链根据输入内容动态选择不同的处理路径，适用于需要多种处理策略的场景。

from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch

llm = ChatOpenAI(model="gpt-4o")

# 不同类型的提示
math_prompt = ChatPromptTemplate.from_template(
    "你是一个数学专家。请解答以下数学问题，给出详细的解题步骤：\n\n{question}"
)

code_prompt = ChatPromptTemplate.from_template(
    "你是一个编程专家。请解答以下编程问题，提供完整的代码示例：\n\n{question}"
)

general_prompt = ChatPromptTemplate.from_template(
    "请回答以下问题：\n\n{question}"
)

# 分类链
classify_prompt = ChatPromptTemplate.from_template(
    "请判断以下问题属于哪个类别，只回答类别名称（math/code/general）：\n\n{question}"
)

# 路由链
def classify(input_dict):
    response = (classify_prompt | llm | StrOutputParser()).invoke(input_dict)
    category = response.strip().lower()
    if "math" in category:
        return "math"
    elif "code" in category:
        return "code"
    return "general"

# 使用RunnableBranch
branch = RunnableBranch(
    (lambda x: classify(x) == "math", math_prompt | llm | StrOutputParser()),
    (lambda x: classify(x) == "code", code_prompt | llm | StrOutputParser()),
    general_prompt | llm | StrOutputParser()  # 默认分支
)

# 测试
result = branch.invoke({"question": "如何用Python实现快速排序？"})
print(result)

4.4 并行处理链

LangChain支持并行执行多个链，然后合并结果。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel

llm = ChatOpenAI(model="gpt-4o")

# 并行执行多个分析
sentiment_chain = (
    ChatPromptTemplate.from_template("分析以下文本的情感倾向（正面/负面/中性）：\n{text}")
    | llm | StrOutputParser()
)

summary_chain = (
    ChatPromptTemplate.from_template("请用一句话总结以下文本：\n{text}")
    | llm | StrOutputParser()
)

keywords_chain = (
    ChatPromptTemplate.from_template("提取以下文本的3-5个关键词：\n{text}")
    | llm | StrOutputParser()
)

# 并行组合
parallel_chain = RunnableParallel(
    sentiment=sentiment_chain,
    summary=summary_chain,
    keywords=keywords_chain
)

# 执行
results = parallel_chain.invoke({
    "text": "LangChain是一个非常强大的框架，它大大简化了LLM应用的开发流程。虽然学习曲线稍陡，但一旦掌握就能显著提高开发效率。"
})

print(f"情感: {results['sentiment']}")
print(f"总结: {results['summary']}")
print(f"关键词: {results['keywords']}")

---

五、Agent与工具使用

5.1 Agent的概念

Agent是LangChain中最具威力的组件之一。与固定的链不同，Agent能够根据用户的输入和当前的上下文动态决定使用哪些工具、以什么顺序使用，以及如何组合结果。它实现了一个"思考-行动-观察"（ReAct）的循环模式。

5.2 创建自定义工具

from langchain_core.tools import tool
from typing import Annotated
import math

# 使用@tool装饰器创建工具
@tool
def calculator(expression: Annotated[str, "数学表达式，例如 '2+3*4' 或 'sqrt(16)'"]) -> str:
    """计算数学表达式的结果。支持基本运算和常用数学函数。"""
    try:
        # 安全的数学表达式求值
        allowed_names = {
            "abs": abs, "round": round,
            "sin": math.sin, "cos": math.cos, "tan": math.tan,
            "sqrt": math.sqrt, "log": math.log, "pi": math.pi, "e": math.e
        }
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return str(result)
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool
def get_current_time() -> str:
    """获取当前的日期和时间。"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def search_knowledge_base(
    query: Annotated[str, "搜索查询"],
    top_k: Annotated[int, "返回结果数量"] = 3
) -> str:
    """在知识库中搜索相关信息。"""
    # 模拟知识库搜索
    knowledge = {
        "Python": "Python是一种高级编程语言，由Guido van Rossum于1991年首次发布。",
        "机器学习": "机器学习是人工智能的一个分支，通过算法让计算机从数据中学习规律。",
        "深度学习": "深度学习是机器学习的子集，使用多层神经网络来学习数据的层次化表示。"
    }
    results = []
    for key, value in knowledge.items():
        if key in query or query in value:
            results.append(f"[{key}] {value}")
    return "\n".join(results[:top_k]) if results else "未找到相关信息。"

print(f"工具列表: {[calculator.name, get_current_time.name, search_knowledge_base.name]}")

5.3 使用create_react_agent创建Agent

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

# 初始化模型
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 工具列表
tools = [calculator, get_current_time, search_knowledge_base]

# 创建ReAct Agent
agent = create_react_agent(llm, tools)

# 调用Agent
response = agent.invoke({
    "messages": [("human", "现在几点了？另外请帮我计算 sqrt(144) + 3*7")]
})

# 输出结果
for message in response["messages"]:
    print(f"[{message.type}] {message.content}")

5.4 自定义Agent行为

from langgraph.prebuilt import create_react_agent
from langchain_core.messages import SystemMessage

# 带系统提示的Agent
system_message = SystemMessage(content="""
你是一个专业的技术助手。在回答问题时：
1. 如果涉及计算，使用calculator工具
2. 如果需要时间信息，使用get_current_time工具
3. 如果需要查询知识，使用search_knowledge_base工具
4. 始终用中文回答
5. 给出答案后，简要说明你是如何得出这个答案的
""")

agent = create_react_agent(
    llm,
    tools,
    state_modifier=system_message
)

response = agent.invoke({
    "messages": [("human", "Python是什么时候发布的？现在距离发布多少年了？")]
})

---

六、记忆系统（Memory System）

6.1 对话记忆的重要性

在构建聊天应用时，记忆系统使AI能够记住之前的对话内容，从而进行连贯的多轮对话。LangChain提供了多种记忆策略。

6.2 基于消息列表的内存管理

在现代LangChain中，推荐使用LangGraph来管理对话状态和记忆：

from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent

llm = ChatOpenAI(model="gpt-4o")

# 创建记忆存储
memory = MemorySaver()

# 创建带记忆的Agent
agent = create_react_agent(
    llm,
    tools=[calculator, get_current_time],
    checkpointer=memory
)

# 配置会话ID
config = {"configurable": {"thread_id": "user-session-123"}}

# 第一轮对话
response1 = agent.invoke({
    "messages": [("human", "我叫张三，我正在学习Python。")]
}, config=config)
print(response1["messages"][-1].content)

# 第二轮对话（Agent会记住之前的上下文）
response2 = agent.invoke({
    "messages": [("human", "我的名字是什么？我在学什么？")]
}, config=config)
print(response2["messages"][-1].content)
# 输出会包含"张三"和"Python"

6.3 对话摘要记忆

对于长对话，完整保存所有消息会消耗大量token。摘要记忆策略会定期将对话历史压缩为摘要。

from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o")

def summarize_conversation(messages: list, max_history: int = 10) -> list:
    """当消息超过阈值时，压缩历史为摘要"""
    if len(messages) <= max_history:
        return messages
    
    # 将旧消息压缩为摘要
    old_messages = messages[:-max_history]
    recent_messages = messages[-max_history:]
    
    summary_prompt = ChatPromptTemplate.from_template(
        "请将以下对话历史压缩为简短摘要，保留关键信息：\n\n{history}"
    )
    
    history_text = "\n".join([
        f"{'用户' if isinstance(m, HumanMessage) else 'AI'}: {m.content}"
        for m in old_messages
    ])
    
    summary_chain = summary_prompt | llm
    summary = summary_chain.invoke({"history": history_text})
    
    # 用摘要替换旧消息
    return [
        HumanMessage(content=f"[对话摘要] {summary.content}")
    ] + recent_messages

6.4 持久化记忆存储

from langgraph.checkpoint.sqlite import SqliteSaver

# 使用SQLite持久化对话记忆
import sqlite3
conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
memory = SqliteSaver(conn)

agent = create_react_agent(
    llm,
    tools=[calculator],
    checkpointer=memory
)

# 即使程序重启，之前的对话记忆仍然可用
config = {"configurable": {"thread_id": "persistent-user-123"}}
response = agent.invoke({
    "messages": [("human", "还记得我之前问的问题吗？")]
}, config=config)

---

七、文档加载器与文本分割器

7.1 文档加载器

LangChain支持从多种来源加载文档，包括PDF、Word、网页、数据库等。

from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    CSVLoader,
    UnstructuredMarkdownLoader,
    WebBaseLoader
)

# 加载PDF文件
pdf_loader = PyPDFLoader("document.pdf")
pdf_docs = pdf_loader.load()
print(f"PDF文档页数: {len(pdf_docs)}")
print(f"第一页内容预览: {pdf_docs[0].page_content[:200]}")

# 加载文本文件
text_loader = TextLoader("data.txt", encoding="utf-8")
text_docs = text_loader.load()

# 加载网页
web_loader = WebBaseLoader("https://docs.python.org/3/tutorial/")
web_docs = web_loader.load()
print(f"网页内容长度: {len(web_docs[0].page_content)}")

# 加载CSV文件
csv_loader = CSVLoader("data.csv", encoding="utf-8")
csv_docs = csv_loader.load()

# 查看文档结构
for doc in pdf_docs[:2]:
    print(f"来源: {doc.metadata.get('source', '未知')}")
    print(f"页码: {doc.metadata.get('page', '未知')}")
    print(f"内容长度: {len(doc.page_content)} 字符")
    print("---")

7.2 文本分割器

文本分割器将长文档切分为较小的块（chunks），这对RAG系统至关重要——因为LLM的上下文窗口有限，且检索需要精确的文本片段。

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    TokenTextSplitter,
    MarkdownHeaderTextSplitter
)

# 最常用：递归字符分割器
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,       # 每个块的最大字符数
    chunk_overlap=200,     # 块之间的重叠字符数
    length_function=len,
    separators=["\n\n", "\n", "。", "！", "？", ".", " ", ""]
)

# 分割文档
chunks = splitter.split_documents(pdf_docs)
print(f"原始文档数: {len(pdf_docs)}")
print(f"分割后块数: {len(chunks)}")
print(f"平均块大小: {sum(len(c.page_content) for c in chunks) / len(chunks):.0f} 字符")

# Markdown标题分割器
markdown_text = """
# 第一章 概述
## 1.1 背景
这是背景内容...

## 1.2 目标
这是目标内容...

# 第二章 技术方案
## 2.1 架构设计
这是架构内容...
"""

headers_to_split = [
    ("#", "一级标题"),
    ("##", "二级标题"),
    ("###", "三级标题"),
]

md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split)
md_chunks = md_splitter.split_text(markdown_text)

for chunk in md_chunks:
    print(f"元数据: {chunk.metadata}")
    print(f"内容: {chunk.page_content[:100]}")
    print("---")

7.3 分割策略选择指南

选择合适的分割策略对检索质量至关重要：

场景	推荐分割器	理由
通用文本	RecursiveCharacterTextSplitter	按语义边界分割，保持语义完整性
代码文件	RecursiveCharacterTextSplitter + 语言专用分隔符	保持代码结构完整
Markdown文档	MarkdownHeaderTextSplitter	按标题层次分割
对话记录	按对话轮次分割	保持对话上下文完整
长文档摘要	TokenTextSplitter	精确控制token数量

---

八、向量存储与检索器

8.1 向量存储基础

向量存储是RAG系统的核心组件，负责存储文档的嵌入向量并支持高效的相似性搜索。

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document

# 准备文档
documents = [
    Document(page_content="Python是一种解释型、面向对象的高级编程语言。", metadata={"source": "python_doc"}),
    Document(page_content="机器学习是人工智能的一个子领域，通过算法从数据中学习。", metadata={"source": "ml_doc"}),
    Document(page_content="深度学习使用多层神经网络来学习数据的层次化表示。", metadata={"source": "dl_doc"}),
    Document(page_content="自然语言处理使计算机能够理解和生成人类语言。", metadata={"source": "nlp_doc"}),
    Document(page_content="计算机视觉让机器能够从图像和视频中提取信息。", metadata={"source": "cv_doc"}),
]

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 创建FAISS向量存储
vectorstore = FAISS.from_documents(documents, embeddings)

# 相似性搜索
results = vectorstore.similarity_search("什么是深度学习？", k=3)
for doc in results:
    print(f"[{doc.metadata['source']}] {doc.page_content}")

8.2 其他向量数据库

# Chroma（轻量级，适合开发和小规模生产）
from langchain_community.vectorstores import Chroma

vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# Pinecone（云服务，适合大规模生产）
from langchain_pinecone import PineconeVectorStore

vectorstore = PineconeVectorStore.from_documents(
    documents=chunks,
    embedding=embeddings,
    index_name="my-index"
)

# Milvus（分布式，适合大规模部署）
from langchain_community.vectorstores import Milvus

vectorstore = Milvus.from_documents(
    documents=chunks,
    embedding=embeddings,
    connection_args={"host": "localhost", "port": 19530}
)

8.3 检索器（Retriever）

检索器是从向量存储中获取相关文档的统一接口，支持多种检索策略。

# 基本检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}
)

# 最大边际相关性检索（MMR）- 在相关性和多样性之间取得平衡
retriever_mmr = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.7}
)

# 带分数阈值的检索
retriever_score = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.7, "k": 5}
)

# 使用检索器
docs = retriever.invoke("机器学习的基本概念")
for doc in docs:
    print(f"来源: {doc.metadata['source']}, 内容: {doc.page_content[:100]}")

8.4 高级检索策略

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

# 上下文压缩检索：使用LLM从检索到的文档中提取与查询相关的部分
compressor = LLMChainExtractor.from_llm(ChatOpenAI(model="gpt-4o"))
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10})
)

# 多查询检索：生成多个查询变体来提高召回率
from langchain.retrievers.multi_query import MultiQueryRetriever

multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=ChatOpenAI(model="gpt-4o")
)

---

九、LangChain Expression Language (LCEL)

9.1 LCEL概述

LCEL是LangChain的核心声明式语言，它通过管道操作符（|）将组件组合成可执行的链。LCEL具有以下特点：

• 流式支持：自动支持stream和ainvoke等异步操作
• 批处理支持：自动支持batch操作
• 并行执行：RunnableParallel支持并行处理
• 错误处理：支持重试和回退机制
• 可追踪性：与LangSmith无缝集成

9.2 核心Runnable接口

from langchain_core.runnables import (
    RunnablePassthrough,
    RunnableLambda,
    RunnableParallel,
    RunnableBranch,
    RunnableConfig
)

# RunnablePassthrough - 直接传递输入
passthrough = RunnablePassthrough()
result = passthrough.invoke("hello")  # 输出: "hello"

# RunnableLambda - 将普通函数包装为Runnable
def process_text(text: str) -> str:
    return text.upper().strip()

processor = RunnableLambda(process_text)
result = processor.invoke("  hello world  ")  # 输出: "HELLO WORLD"

# 链式组合
chain = passthrough | processor
result = chain.invoke("  test  ")  # 输出: "TEST"

9.3 输入输出与数据传递

from langchain_core.runnables import RunnablePassthrough, RunnableLambda

# 构建RAG链
def format_docs(docs):
    """将检索到的文档格式化为字符串"""
    return "\n\n".join(doc.page_content for doc in docs)

# RAG链
rag_prompt = ChatPromptTemplate.from_template("""
请根据以下上下文回答问题。如果上下文中没有相关信息，请说明无法回答。

上下文：
{context}

问题：{question}

回答：
""")

rag_chain = (
    {
        "context": retriever | RunnableLambda(format_docs),
        "question": RunnablePassthrough()
    }
    | rag_prompt
    | llm
    | StrOutputParser()
)

# 使用RAG链
answer = rag_chain.invoke("什么是深度学习？")
print(answer)

9.4 流式输出

# 同步流式输出
for chunk in rag_chain.stream("什么是机器学习？"):
    print(chunk, end="", flush=True)
print()

# 异步调用
import asyncio

async def async_invoke():
    result = await rag_chain.ainvoke("什么是自然语言处理？")
    return result

result = asyncio.run(async_invoke())

9.5 重试与回退机制

from langchain_core.runnables import RunnableWithFallbacks

# 定义回退模型
primary_llm = ChatOpenAI(model="gpt-4o")
fallback_llm = ChatOpenAI(model="gpt-3.5-turbo")

# 使用回退机制
chain_with_fallback = primary_llm.with_fallbacks([fallback_llm])

# 重试机制
from langchain_core.runnables import RunnableRetry

chain_with_retry = rag_chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True
)

---

十、LangSmith调试与监控

10.1 LangSmith简介

LangSmith是LangChain官方提供的调试、追踪、评估和监控平台。它能够记录每次LLM调用的详细信息，帮助开发者快速定位问题。

10.2 配置与使用

import os

# 启用LangSmith追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-rag-project"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"

# 配置完成后，所有LangChain调用都会自动被追踪
# 无需修改任何业务代码
chain = rag_prompt | llm | StrOutputParser()
result = chain.invoke({"context": "...", "question": "..."})
# 在LangSmith UI中可以查看完整的调用链路

10.3 评估与测试

from langsmith import Client
from langchain.smith import RunEvalConfig

client = Client()

# 定义评估数据集
dataset_name = "rag-evaluation"
examples = [
    {"input": "什么是深度学习？", "output": "深度学习是机器学习的一个子领域..."},
    {"input": "Python有什么特点？", "output": "Python是一种解释型、面向对象的编程语言..."},
]

# 创建评估数据集
dataset = client.create_dataset(dataset_name)
for example in examples:
    client.create_example(
        inputs={"question": example["input"]},
        outputs={"answer": example["output"]},
        dataset_id=dataset.id
    )

# 运行评估
eval_config = RunEvalConfig(
    evaluators=[
        "qa",  # QA评估器
        "criteria": {"relevance": "回答是否与问题相关？"}
    ]
)

---

十一、与外部API集成

11.1 工具与API集成模式

from langchain_core.tools import tool
import httpx

@tool
async def fetch_weather(city: str) -> str:
    """查询指定城市的天气信息。"""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.openweathermap.org/data/2.5/weather",
            params={"q": city, "appid": "your-api-key", "units": "metric", "lang": "zh_cn"}
        )
        data = response.json()
        temp = data["main"]["temp"]
        desc = data["weather"][0]["description"]
        return f"{city}当前天气：{desc}，温度{temp}°C"

@tool
async def search_web(query: str) -> str:
    """在互联网上搜索信息。"""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.search.com/search",
            params={"q": query, "limit": 5}
        )
        results = response.json().get("results", [])
        return "\n".join([f"- {r['title']}: {r['snippet']}" for r in results])

11.2 与数据库集成

from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import create_sql_agent

# 连接数据库
db = SQLDatabase.from_uri("sqlite:///my_database.db")

# 创建SQL Agent
sql_agent = create_sql_agent(
    llm=ChatOpenAI(model="gpt-4o"),
    db=db,
    agent_type="openai-tools",
    verbose=True
)

# 使用自然语言查询数据库
response = sql_agent.invoke({
    "input": "查询最近7天注册的用户数量，按天分组"
})
print(response["output"])

---

十二、实战案例：构建智能客服系统

12.1 系统架构

我们将构建一个完整的智能客服系统，具备以下功能：

• 多轮对话支持
• 知识库检索
• 工具调用（查询订单、创建工单等）
• 对话历史管理
• 优雅的降级处理

12.2 完整实现

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.tools import tool
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent
from typing import Annotated

# ==================== 1. 知识库准备 ====================

# 模拟客服知识库文档
knowledge_base_docs = [
    """退换货政策：自收到商品之日起7天内，如商品保持原样且未使用，可申请无理由退换货。
    退换货流程：1.登录账户 2.进入订单详情 3.点击"申请退换货" 4.填写原因 5.等待审核（1-2个工作日）。
    特殊商品（如贴身衣物、食品等）不支持无理由退换。""",
    
    """配送政策：标准配送3-5个工作日到达，加急配送1-2个工作日到达。
    订单金额满99元免运费，不满99元收取8元运费。
    偏远地区（新疆、西藏等）可能需要额外3-5个工作日。""",
    
    """会员权益：
    普通会员：积分1倍，无专属折扣
    银卡会员：积分1.5倍，95折优惠
    金卡会员：积分2倍，9折优惠，免费退换货
    钻石会员：积分3倍，85折优惠，免费退换货，专属客服""",
    
    """支付方式：支持微信支付、支付宝、银行卡（借记卡/信用卡）、花呗分期。
    分期付款：3期免息，6期手续费2.5%，12期手续费4.5%。
    退款将在审核通过后3-5个工作日内原路退回。"""
]

# 创建向量存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
all_chunks = []
for doc in knowledge_base_docs:
    chunks = text_splitter.create_documents([doc])
    all_chunks.extend(chunks)

vectorstore = FAISS.from_documents(all_chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# ==================== 2. 定义工具 ====================

@tool
def query_knowledge_base(
    query: Annotated[str, "用户的问题或查询关键词"]
) -> str:
    """从客服知识库中检索相关信息。用于回答关于政策、流程、权益等问题。"""
    docs = retriever.invoke(query)
    if docs:
        return "\n\n".join([doc.page_content for doc in docs])
    return "未找到相关信息，建议联系人工客服。"

@tool
def query_order_status(
    order_id: Annotated[str, "订单号"]
) -> str:
    """查询订单状态。输入订单号，返回订单的当前状态、物流信息等。"""
    # 模拟订单查询
    mock_orders = {
        "ORD2024001": {"status": "已发货", "tracking": "SF1234567890", "eta": "2024-01-15"},
        "ORD2024002": {"status": "待付款", "tracking": "", "eta": ""},
        "ORD2024003": {"status": "已完成", "tracking": "SF9876543210", "eta": "已签收"},
    }
    order = mock_orders.get(order_id)
    if order:
        return f"订单 {order_id} 状态: {order['status']}\n物流单号: {order['tracking']}\n预计送达: {order['eta']}"
    return f"未找到订单 {order_id}，请检查订单号是否正确。"

@tool
def create_support_ticket(
    user_id: Annotated[str, "用户ID"],
    issue_type: Annotated[str, "问题类型：退换货/配送/支付/其他"],
    description: Annotated[str, "问题描述"]
) -> str:
    """创建客服工单。当用户的问题无法通过知识库解决时使用。"""
    # 模拟创建工单
    ticket_id = f"TK{hash(user_id + description) % 100000:05d}"
    return f"工单已创建！\n工单号: {ticket_id}\n问题类型: {issue_type}\n描述: {description}\n预计处理时间: 24小时内，我们会尽快为您处理。"

@tool
def transfer_to_human(
    reason: Annotated[str, "转接人工客服的原因"]
) -> str:
    """将对话转接给人工客服。当用户明确要求人工客服或问题超出AI处理能力时使用。"""
    return f"正在为您转接人工客服，请稍候...\n转接原因: {reason}\n预计等待时间: 2-5分钟"

# ==================== 3. 创建智能客服Agent ====================

llm = ChatOpenAI(model="gpt-4o", temperature=0.3)

# 系统提示
system_prompt = """你是一位专业、友善的电商客服助手。请遵循以下原则：

1. 始终保持礼貌和耐心，使用亲切的语气
2. 优先使用知识库工具查询相关信息回答用户问题
3. 对于订单相关问题，使用订单查询工具
4. 如果问题无法解决，主动创建工单或转接人工客服
5. 回答要简洁明了，避免冗长的技术术语
6. 如果用户情绪激动，先表示理解和歉意，再解决问题
7. 保护用户隐私，不要主动索要敏感信息

你可以使用的工具：
- query_knowledge_base: 查询退换货、配送、会员权益、支付等政策信息
- query_order_status: 查询订单状态和物流信息
- create_support_ticket: 创建客服工单
- transfer_to_human: 转接人工客服"""

# 记忆存储
memory = MemorySaver()

# 创建Agent
customer_service_agent = create_react_agent(
    llm,
    tools=[query_knowledge_base, query_order_status, create_support_ticket, transfer_to_human],
    checkpointer=memory,
    state_modifier=SystemMessage(content=system_prompt)
)

# ==================== 4. 对话交互 ====================

def chat(user_input: str, session_id: str = "default") -> str:
    """与客服Agent交互"""
    config = {"configurable": {"thread_id": session_id}}
    response = customer_service_agent.invoke(
        {"messages": [HumanMessage(content=user_input)]},
        config=config
    )
    return response["messages"][-1].content

# 模拟多轮对话
print("=== 智能客服系统启动 ===\n")

# 对话1：查询退换货政策
response1 = chat("你好，我买的衣服不太合适，想退掉，可以吗？", "session-001")
print(f"客服: {response1}\n")

# 对话2：查询订单状态
response2 = chat("我的订单号是ORD2024001，帮我查一下到哪了？", "session-001")
print(f"客服: {response2}\n")

# 对话3：需要人工帮助
response3 = chat("这个订单的物流信息好像有问题，我需要人工客服帮我处理。", "session-001")
print(f"客服: {response3}\n")

12.3 系统优化建议

1. 性能优化：对高频查询结果进行缓存，减少向量搜索次数
2. 监控告警：集成LangSmith监控，设置异常告警规则
3. A/B测试：对不同的系统提示和检索策略进行A/B测试
4. 用户反馈：收集用户满意度反馈，持续优化模型表现
5. 安全防护：添加输入过滤，防止提示注入攻击

---

十三、最佳实践

13.1 提示工程最佳实践

# 好的系统提示：明确、具体、有约束
system_prompt = """你是一个技术支持助手。

规则：
1. 只回答与产品技术问题相关的问题
2. 如果不确定答案，坦诚告知并建议联系技术支持
3. 回答格式：先给出结论，再给出详细解释
4. 代码示例使用Python 3.10+
5. 每次回答控制在500字以内

输出格式：
## 结论
[简短结论]

## 详细说明
[详细解释]

## 代码示例（如适用）
```python
[代码]
```"""

13.2 错误处理

from langchain_core.runnables import RunnableConfig

def safe_invoke(chain, input_data, config: RunnableConfig = None):
    """安全调用链，包含错误处理和日志记录"""
    try:
        result = chain.invoke(input_data, config)
        return {"success": True, "result": result}
    except Exception as e:
        error_msg = str(e)
        if "rate_limit" in error_msg.lower():
            return {"success": False, "error": "API调用频率超限，请稍后重试"}
        elif "context_length" in error_msg.lower():
            return {"success": False, "error": "输入内容过长，请缩短输入"}
        else:
            return {"success": False, "error": f"处理失败: {error_msg}"}

13.3 生产部署注意事项

1. API密钥管理：使用环境变量或密钥管理服务，不要硬编码
2. 速率限制：实现请求队列和重试机制，避免API限流
3. 成本控制：监控token使用量，设置预算告警
4. 日志记录：记录所有LLM调用的输入输出，便于调试和审计
5. 数据安全：对敏感数据进行脱敏处理后再发送给LLM
6. 版本锁定：锁定依赖版本，避免框架升级导致兼容性问题

---

十四、常见问题解答

Q1: 如何选择合适的Chat Model？

选择模型时需要考虑以下因素：

• 性能需求：GPT-4o和Claude-3.5-Sonnet适合复杂任务，GPT-3.5-turbo适合简单任务
• 成本预算：按token计费，小模型成本更低
• 延迟要求：小模型响应更快
• 特殊能力：如需要视觉能力，选择支持多模态的模型

Q2: chunk_size应该设置多大？

chunk_size的选择取决于多个因素：

• 模型上下文窗口：通常设置为模型上下文窗口的1/4到1/2
• 检索精度：较小的chunk_size提高检索精度，但可能丢失上下文
• 典型值：500-1500字符是常见的选择范围
• 建议：通过实验确定最佳值，使用评估指标（如准确率、召回率）来衡量

Q3: 如何处理多语言场景？

# 使用支持多语言的嵌入模型
multilingual_embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small"  # 支持多语言
)

# 在系统提示中指定语言
system_prompt = "请使用用户使用的语言回答问题。如果用户用中文提问，用中文回答；用英文提问，用英文回答。"

Q4: LangChain和LangGraph有什么区别？

• LangChain：提供基础组件和链式调用模式，适合简单的线性流程
• LangGraph：基于图的有状态编排框架，适合复杂的、需要条件分支和循环的Agent场景
• 建议：新项目推荐使用LangGraph构建Agent，它更灵活且支持更复杂的控制流

Q5: 如何降低LLM调用成本？

1. 使用更小的模型处理简单任务（路由、分类等）
2. 实施缓存策略，对相同或相似查询返回缓存结果
3. 优化prompt，减少不必要的token消耗
4. 使用流式输出减少用户感知延迟
5. 批量处理请求，减少API调用次数

---

十五、总结

LangChain作为构建LLM应用的主流框架，提供了丰富且灵活的组件体系。本教程涵盖了从基础概念到高级应用的完整知识体系：

1. 模型接口：统一了不同LLM提供商的调用方式
2. Prompt工程：提供了模板化和结构化的提示管理方案
3. 链式组合：通过LCEL实现了声明式的组件编排
4. Agent系统：实现了动态工具选择和自主决策
5. 记忆管理：支持多种对话记忆策略
6. RAG系统：提供了完整的文档加载、分割、嵌入、检索流程
7. 调试监控：通过LangSmith实现了全链路追踪

掌握这些核心概念和最佳实践，你就能构建出功能强大、稳定可靠的LLM应用。随着技术的不断发展，LangChain也在持续演进，建议开发者关注官方文档和社区动态，及时了解最新的特性和改进。

下一步学习建议：

• 深入学习LangGraph，掌握复杂Agent的构建
• 探索LangSmith的高级评估功能
• 研究生产环境的部署和运维最佳实践
• 参与LangChain开源社区，贡献代码和反馈

---

本教程基于LangChain v0.3+版本编写，部分API可能随版本更新而变化，请以官方文档为准。