LangChain 技术详解与应用实战

zhaoshuzhaoshu

397人浏览 · 2026-05-28 13:04:54

zhaoshuzhaoshu · 2026-05-28 13:04:54 发布

本文档系统介绍 LangChain 框架的核心技术、架构设计、典型应用场景及完整案例分析，并附带可直接运行的 Python示例

1. LangChain 概述

1.1 什么是 LangChain

LangChain 是一个基于 Python 的开源大语言模型（LLM）应用开发框架，旨在简化 LLM 驱动的应用程序构建流程。它提供标准化的接口来连接各种大语言模型、嵌入模型、向量数据库等组件，让开发者能够像搭积木一样快速组装复杂的 AI 应用。

LangChain 的核心价值在于：

模块化设计：将 LLM 应用拆分为可复用的组件
标准化接口：统一不同模型和工具的调用方式
生态丰富：拥有最大的社区和最多的第三方集成
生产就绪：提供从开发到部署的全生命周期支持

1.2 发展现状（2026年）

截至 2026 年，LangChain 已成为生态最完善的 Agent 框架，拥有 97,000+ GitHub Star 和 50,000+ 生产应用。2026 年的重大更新包括：

LangGraph 成为 Agent 编排核心：支持复杂的状态机和循环逻辑
LangSmith Fleet 发布：原 Agent Builder 升级，提供更专业的智能体管理与部署能力
NVIDIA 深度集成：优化底层算力支持
流式元数据兼容性增强：更高效地处理动态数据流

2. 核心架构与技术体系

2.1 技术体系全景

LangChain 采用分层模块化设计，主要包含以下核心包：

模块	说明	职责
`langchain-core`	基础核心包	封装聊天模型、消息类型、输出解析器等基础组件
`langchain`	主框架包	提供 Chain、Agent、Retriever 等高层抽象
`langchain-openai` / `langchain-anthropic` 等	官方集成包	与各大 LLM 提供商的深度集成
`langchain-community`	社区集成包	社区维护的第三方工具集成
`langgraph`	图编排引擎	用"图"管理复杂任务流程，支持循环和分支
`langsmith`	全生命周期管理平台	覆盖开发→测试→部署→监控全流程

2.2 核心设计哲学

LangChain 以**"链式编排"**为核心设计理念：

组件化：每个功能单元（模型、提示、解析器）都是独立组件
可组合：通过管道符 | 将组件串联成执行链
可观测：每个步骤都可追踪、可调试
可扩展：轻松接入自定义工具和模型

3. 六大核心模块详解

3.1 Model I/O（模型输入输出）

Model I/O 是 LangChain 与 LLM 交互的基础层，包含三个子组件：

3.1.1 Prompts（提示模板）

提示模板用于结构化输入，支持变量插值和少样本学习。

from langchain_core.prompts import ChatPromptTemplate, FewShotPromptTemplate, PromptTemplate

# 基础提示模板
basic_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位专业的{role}。"),
    ("human", "请帮我{task}")
])
messages = basic_prompt.format_messages(role="Python工程师", task="写一个快速排序算法")

# 少样本提示模板
examples = [
    {"input": "飞行员", "output": "飞机"},
    {"input": "驾驶员", "output": "汽车"},
    {"input": "厨师", "output": "厨房"},
]
example_prompt = PromptTemplate(
    input_variables=["input", "output"],
    template="示例输入: {input}, 示例输出: {output}",
)
few_shot_prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    prefix="根据下面示例，写出输出",
    suffix="输入: {value}，输出:",
    input_variables=["value"],
)
print(few_shot_prompt.format(value="学生"))

3.1.2 Language Models（语言模型）

LangChain 支持多种 LLM，通过统一接口调用：

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_anthropic import ChatAnthropic

# OpenAI GPT 模型
openai_model = ChatOpenAI(model="gpt-4o", temperature=0.7, max_tokens=2000)

# Anthropic Claude 模型
claude_model = ChatAnthropic(model="claude-3-sonnet-20240229", temperature=0.5)

# 统一调用接口
response = openai_model.invoke("你好，请介绍一下LangChain框架")
print(response.content)

3.1.3 Output Parsers（输出解析器）

将 LLM 的自由文本输出转换为结构化数据：

from langchain_core.output_parsers import JsonOutputParser, PydanticOutputParser
from pydantic import BaseModel, Field

class MovieReview(BaseModel):
    title: str = Field(description="电影名称")
    rating: int = Field(description="评分（1-10）")
    summary: str = Field(description="简短评价")
    tags: list = Field(description="标签列表")

pydantic_parser = PydanticOutputParser(pydantic_object=MovieReview)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的电影评论家。请按以下格式输出评价：\n{format_instructions}"),
    ("human", "请评价电影《{movie_name}》")
])
prompt_with_format = prompt.partial(format_instructions=pydantic_parser.get_format_instructions())

chain = prompt_with_format | openai_model | pydantic_parser
result = chain.invoke({"movie_name": "星际穿越"})
print(result)

3.2 Retrieval（检索增强）

RAG（Retrieval-Augmented Generation）是 LangChain 的强项，核心流程包括：文档加载 → 文本分割 → 向量化 → 向量存储 → 相似度检索。

from langchain_community.document_loaders import TextLoader, PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 1. 文档加载
text_loader = TextLoader("./docs/article.txt", encoding="utf-8")
pdf_loader = PyPDFLoader("./docs/report.pdf")
dir_loader = DirectoryLoader("./docs/", glob="**/*.txt")
all_docs = dir_loader.load()

# 2. 文本分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", "。", "！", "？", " ", ""]
)
split_docs = text_splitter.split_documents(all_docs)

# 3. 向量化与存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = Chroma.from_documents(
    documents=split_docs,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 4. 检索器
retriever = vectorstore.as_retriever(search_type="similarity", search_kwargs={"k": 5})
relevant_docs = retriever.invoke("LangChain的核心优势是什么？")

3.3 Chains（链）

Chain 是将多个组件串联成执行流程的核心机制。

from langchain.chains import LLMChain, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain

# 基础 LLMChain
prompt = ChatPromptTemplate.from_template("请用一句话总结以下内容：\n\n{content}")
llm_chain = LLMChain(llm=openai_model, prompt=prompt)
result = llm_chain.invoke({"content": "LangChain是一个开源框架..."})

# RAG 问答链
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的问答助手。请基于以下上下文回答问题：\n\n{context}"),
    ("human", "{input}")
])
document_chain = create_stuff_documents_chain(openai_model, rag_prompt)
retrieval_chain = create_retrieval_chain(retriever, document_chain)
response = retrieval_chain.invoke({"input": "LangChain支持哪些向量数据库？"})
print(response["answer"])

3.4 Memory（记忆系统）

Memory 让 AI 应用具备"记住"对话历史的能力。

from langchain.memory import (
    ConversationBufferMemory,
    ConversationBufferWindowMemory,
    ConversationSummaryMemory,
    VectorStoreRetrieverMemory
)
from langchain.chains import ConversationChain

# 缓冲记忆（保存全部历史）
buffer_memory = ConversationBufferMemory(memory_key="history", return_messages=True)
conversation = ConversationChain(llm=openai_model, memory=buffer_memory, verbose=True)
conversation.predict(input="你好，我叫张三")
conversation.predict(input="你还记得我的名字吗？")

# 窗口记忆（只保留最近N轮）
window_memory = ConversationBufferWindowMemory(k=3, memory_key="history")

# 摘要记忆（自动压缩历史）
summary_memory = ConversationSummaryMemory(llm=openai_model, memory_key="history")

# 向量检索记忆（基于语义检索历史）
vector_memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
)

3.5 Agents（智能体）

Agent 是让 LLM 自主决策何时调用工具的机制。

from langchain.agents import create_react_agent, create_openai_tools_agent, AgentExecutor, Tool
from langchain import hub
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

search_tool = DuckDuckGoSearchRun()
wikipedia_tool = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

tools = [
    Tool(name="WebSearch", func=search_tool.run, description="用于搜索互联网上的实时信息"),
    Tool(name="Wikipedia", func=wikipedia_tool.run, description="用于查询维基百科上的知识"),
    Tool(name="Calculator", func=lambda x: str(eval(x)), description="用于执行数学计算"),
]

# ReAct Agent
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(openai_model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = agent_executor.invoke({"input": "2026年LangChain的最新版本是多少？"})

# OpenAI Tools Agent（推荐用于支持 Function Calling 的模型）
tools_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手，可以使用工具来回答问题。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])
openai_agent = create_openai_tools_agent(openai_model, tools, tools_prompt)
openai_executor = AgentExecutor(agent=openai_agent, tools=tools, verbose=True)

3.6 Callbacks（回调系统）

用于监控、日志记录和流式输出。

from langchain_core.callbacks import BaseCallbackHandler
import time

class TimingCallbackHandler(BaseCallbackHandler):
    '''自定义回调：记录每个步骤的执行时间'''

    def on_llm_start(self, serialized, prompts, **kwargs):
        self.start_time = time.time()
        print(f"[LLM开始] 模型: {serialized.get('name', 'unknown')}")

    def on_llm_end(self, response, **kwargs):
        elapsed = time.time() - self.start_time
        print(f"[LLM结束] 耗时: {elapsed:.2f}s")

    def on_chain_start(self, serialized, inputs, **kwargs):
        print(f"[Chain开始] 类型: {serialized.get('name', 'unknown')}")

    def on_chain_end(self, outputs, **kwargs):
        print(f"[Chain结束]")

chain_with_callback = prompt | openai_model
result = chain_with_callback.invoke(
    {"content": "LangChain框架介绍"},
    config={"callbacks": [TimingCallbackHandler()]}
)

# 流式输出
streaming_model = ChatOpenAI(model="gpt-4o", streaming=True)
streaming_chain = prompt | streaming_model
for chunk in streaming_chain.stream({"content": "LangChain的应用场景"}):
    print(chunk.content, end="", flush=True)

4. LCEL：链式表达式语言

LCEL（LangChain Expression Language）是 LangChain 现代化的链式构建语法，使用管道符 | 替代传统的链式类。

4.1 基础语法

from langchain_core.runnables import RunnableLambda, RunnableParallel, RunnableBranch

# 基础管道
chain = prompt | openai_model | StrOutputParser()
result = chain.invoke({"content": "LangChain是什么？"})

# 自定义 Runnable
def add_prefix(text: str) -> str:
    return f"[前缀] {text}"

runnable_chain = RunnableLambda(add_prefix) | prompt | openai_model | StrOutputParser()

# 并行执行
parallel_chain = RunnableParallel({
    "summary": summary_prompt | openai_model | StrOutputParser(),
    "translation": translation_prompt | openai_model | StrOutputParser(),
    "keywords": keyword_prompt | openai_model | StrOutputParser()
})
result = parallel_chain.invoke({"text": "LangChain是一个开源框架..."})

# 条件分支
branch = RunnableBranch(
    (lambda x: "代码" in x["topic"], code_prompt | openai_model),
    (lambda x: "数据" in x["topic"], data_prompt | openai_model),
    default_prompt | openai_model
)

4.2 LCEL 的优势

特性	传统 Chain	LCEL
语法	类继承，配置复杂	管道符 `\|`，简洁直观
流式支持	需额外配置	原生支持
并行执行	难以实现	`RunnableParallel` 一行搞定
类型安全	弱	强类型提示
可观测性	有限	内置追踪

5. LangGraph 与 Agent 编排

5.1 LangGraph 核心概念

LangGraph 专门用于构建能处理复杂任务流程的 AI 应用，核心思想是用**"图"（Graph）**来管理任务流程：

State（状态）：整个图共享的数据对象
Node（节点）：执行具体逻辑的函数
Edge（边）：节点间的连接，决定执行流向
Conditional Edge（条件边）：根据状态动态决定下一步

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    query: str
    search_results: list
    answer: str
    next_step: str

def search_node(state: AgentState):
    query = state["query"]
    results = search_tool.invoke(query)
    return {"search_results": [results], "messages": [("assistant", f"搜索完成: {query}")]}

def analyze_node(state: AgentState):
    results = state["search_results"]
    prompt_text = f"基于以下搜索结果回答问题：\n{results}\n\n问题: {state['query']}"
    response = openai_model.invoke(prompt_text)
    return {"answer": response.content, "messages": [("assistant", "分析完成")]}

def decide_next_step(state: AgentState):
    query = state["query"]
    if any(op in query for op in ["+", "-", "*", "/", "计算"]):
        return {"next_step": "calculate"}
    return {"next_step": "search"}

def calculate_node(state: AgentState):
    query = state["query"]
    try:
        result = eval(query.replace("计算", "").strip())
        return {"answer": str(result), "messages": [("assistant", f"计算结果: {result}")]}
    except:
        return {"next_step": "search"}

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("decide", decide_next_step)
workflow.add_node("search", search_node)
workflow.add_node("analyze", analyze_node)
workflow.add_node("calculate", calculate_node)
workflow.set_entry_point("decide")
workflow.add_conditional_edges(
    "decide",
    lambda state: state["next_step"],
    {"search": "search", "calculate": "calculate"}
)
workflow.add_edge("search", "analyze")
workflow.add_edge("analyze", END)
workflow.add_edge("calculate", END)

app = workflow.compile()
result = app.invoke({
    "query": "2026年LangChain的最新动态",
    "messages": [], "search_results": [], "answer": "", "next_step": ""
})
print(result["answer"])

5.2 LangGraph 适用场景

场景	说明
多步骤审批流程	文档审核、代码审查等需要多轮确认的场景
循环迭代优化	写作助手反复修改直到满足要求
人机协同	需要人工确认关键决策的 Agent
多 Agent 协作	多个专业 Agent 分工完成复杂任务
状态机工作流	有明确状态转换规则的业务流程

6. 典型应用场景

6.1 场景总览

场景	适用度	核心技术	典型行业
RAG 智能问答	★★★★★	Retrieval + Chain	金融、法律、医疗、客服
多步骤工作流	★★★★	Chain + LCEL	电商、内容生成
工具调用 Agent	★★★★	Agent + Tools	数据分析、自动化办公
智能客服	★★★★★	RAG + Memory + Agent	电商、SaaS、金融
文档智能处理	★★★★★	Document Loader + RAG	法律、医疗、企业知识管理
个性化推荐	★★★	Embedding + Retriever	电商、内容平台
代码辅助	★★★★	Agent + Tools	软件开发

6.2 场景一：企业知识库问答系统

业务需求：基于企业内部文档（产品手册、技术文档、规章制度）构建智能问答系统，员工可随时提问获取准确答案。

技术方案：

Document Loader 加载各类文档
Text Splitter 智能分割
Embedding 向量化
Vector Store 存储
RAG Chain 问答

6.3 场景二：智能客服机器人

业务需求：7×24 小时自动回答客户咨询，处理退换货、产品咨询、技术支持等常见问题。

技术方案：

RAG 检索企业知识库
Memory 维持多轮对话上下文
Agent 调用订单查询、物流追踪等工具
情感分析识别用户情绪

6.4 场景三：数据分析与报告生成

业务需求：自动分析销售数据、生成可视化报告、提供业务洞察。

技术方案：

Agent 调用 Pandas 进行数据分析
LLM 生成自然语言解读
工具调用生成图表
输出结构化报告

6.5 场景四：多 Agent 协作系统

业务需求：复杂任务需要多个专业 Agent 协作，如市场调研 = 搜索 Agent + 分析 Agent + 写作 Agent。

技术方案：

LangGraph 编排多 Agent 工作流
每个 Agent 有专门的角色和工具
状态共享和任务传递

7. 完整案例分析：企业知识库问答系统

7.1 项目背景

某科技公司积累了大量内部文档，包括：

产品技术手册（PDF 格式）
开发规范文档（Markdown 格式）
运维操作指南（Word 格式）
常见问题 FAQ（文本格式）

员工日常需要频繁查询这些文档，但传统搜索方式效率低下。公司希望构建一个智能问答系统，让员工用自然语言提问即可获取精准答案。

7.2 系统架构

+-------------------+     +-------------------+     +-------------------+
|   用户提问接口     | --> |   意图理解模块     | --> |   向量检索模块     |
+-------------------+     +-------------------+     +-------------------+
                                                          |
                                                          v
+-------------------+     +-------------------+     +-------------------+
|   答案输出接口     | <-- |   答案生成模块     | <-- |   文档召回模块     |
+-------------------+     +-------------------+     +-------------------+
        ^
        |
+-------------------+
|   对话记忆模块     | (维持多轮上下文)
+-------------------+

7.3 核心流程

阶段一：知识库构建

文档加载：使用 DirectoryLoader 批量读取 ./docs/ 目录下的所有文档，自动识别 PDF、Word、TXT、Markdown 等格式
文本分割：采用 RecursiveCharacterTextSplitter，按段落和句子边界智能切分，设置 chunk_size=1000、chunk_overlap=200，保证语义连贯性
向量化：调用 OpenAIEmbeddings（text-embedding-3-large 模型）将文本块转为 3072 维向量
向量存储：写入 Chroma 向量数据库，设置 persist_directory 实现持久化

阶段二：问答链路

用户提问接收：通过 Web 接口或命令行接收自然语言问题
语义检索：将用户问题向量化，在 Chroma 中执行相似度搜索，召回 top-5 最相关文档块
上下文组装：将召回的文档块按相关性排序，拼接成上下文字符串
提示构建：将系统角色设定 + 上下文 + 用户问题组装成完整提示
LLM 生成：调用 GPT-4o 生成答案，temperature 设为 0.3 保证回答稳定准确
结果返回：将生成的答案返回给用户，同时记录到对话历史

阶段三：记忆管理

对话历史存储：使用 ConversationBufferMemory 保存完整对话记录
上下文注入：每次新提问时，将历史对话作为上下文注入提示模板
历史压缩：当对话轮数超过 10 轮时，自动触发 ConversationSummaryMemory 进行摘要压缩

7.4 完整代码实现

'''
企业知识库智能问答系统 - 基于 LangChain 完整实现
功能：文档加载、向量化存储、语义检索、多轮对话问答
'''

import os
from typing import List
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import (
    PyPDFLoader, 
    DirectoryLoader, 
    TextLoader,
    UnstructuredMarkdownLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.memory import ConversationBufferMemory
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser


# ========== 配置 ==========
LLM_MODEL = "gpt-4o"
EMBEDDING_MODEL = "text-embedding-3-large"
CHUNK_SIZE = 1000
CHUNK_OVERLAP = 200
TOP_K = 5
PERSIST_DIR = "./kb_chroma_db"

llm = ChatOpenAI(model=LLM_MODEL, temperature=0.3)
embeddings = OpenAIEmbeddings(model=EMBEDDING_MODEL)


# ========== 阶段一：知识库构建 ==========
class KnowledgeBaseBuilder:
    '''知识库构建器'''

    def __init__(self, doc_dir: str = "./docs"):
        self.doc_dir = doc_dir
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=CHUNK_SIZE,
            chunk_overlap=CHUNK_OVERLAP,
            separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
        )

    def load_all_documents(self) -> List:
        '''批量加载多种格式文档'''
        documents = []

        # 加载 PDF
        pdf_loader = DirectoryLoader(
            self.doc_dir, glob="**/*.pdf", loader_cls=PyPDFLoader
        )
        documents.extend(pdf_loader.load())

        # 加载 Markdown
        md_loader = DirectoryLoader(
            self.doc_dir, glob="**/*.md", loader_cls=UnstructuredMarkdownLoader
        )
        documents.extend(md_loader.load())

        # 加载纯文本
        txt_loader = DirectoryLoader(
            self.doc_dir, glob="**/*.txt", loader_cls=TextLoader
        )
        documents.extend(txt_loader.load())

        print(f"共加载 {len(documents)} 个原始文档")
        return documents

    def build(self) -> Chroma:
        '''构建并持久化向量知识库'''
        documents = self.load_all_documents()
        chunks = self.text_splitter.split_documents(documents)
        print(f"分割为 {len(chunks)} 个文本块")

        vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=embeddings,
            persist_directory=PERSIST_DIR
        )
        print(f"知识库已保存至 {PERSIST_DIR}")
        return vectorstore


# ========== 阶段二：问答引擎 ==========
class QASystem:
    '''智能问答系统'''

    def __init__(self, vectorstore: Chroma = None):
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )

        if vectorstore is None:
            vectorstore = Chroma(
                persist_directory=PERSIST_DIR,
                embedding_function=embeddings
            )

        self.retriever = vectorstore.as_retriever(
            search_type="similarity",
            search_kwargs={"k": TOP_K}
        )

        # 构建 RAG 链
        self.qa_chain = self._build_chain()

    def _build_chain(self):
        '''构建 LCEL 风格的问答链'''

        # 系统提示模板
        system_prompt = '''你是一个专业的企业知识库问答助手。
        请严格基于以下检索到的上下文内容回答问题。
        如果上下文中没有相关信息，请明确告知"根据现有资料，无法回答该问题"。
        回答应简洁准确，必要时可以分点说明。

        历史对话记录：
        {chat_history}

        检索到的相关文档：
        {context}

        当前问题：{input}
        '''

        prompt = ChatPromptTemplate.from_template(system_prompt)

        # 文档格式化
        def format_docs(docs):
            return "\n\n---\n\n".join([
                f"[文档 {i+1}]\n{doc.page_content}"
                for i, doc in enumerate(docs)
            ])

        # LCEL 链：检索 -> 格式化 -> 注入提示 -> LLM生成 -> 解析输出
        chain = (
            {
                "context": self.retriever | format_docs,
                "chat_history": lambda x: self.memory.load_memory_variables({})["chat_history"],
                "input": RunnablePassthrough()
            }
            | prompt
            | llm
            | StrOutputParser()
        )

        return chain

    def ask(self, question: str) -> str:
        '''提问并返回答案'''
        answer = self.qa_chain.invoke(question)

        # 保存到记忆
        self.memory.save_context(
            {"input": question},
            {"output": answer}
        )

        return answer

    def chat(self, question: str) -> dict:
        '''完整对话（包含检索详情）'''
        # 先检索看召回内容
        retrieved_docs = self.retriever.invoke(question)

        # 生成答案
        answer = self.ask(question)

        return {
            "question": question,
            "answer": answer,
            "retrieved_sources": [
                {
                    "content": doc.page_content[:200] + "...",
                    "metadata": doc.metadata
                }
                for doc in retrieved_docs
            ],
            "history_length": len(self.memory.chat_memory.messages)
        }


# ========== 阶段三：运行示例 ==========
if __name__ == "__main__":
    # 首次运行：构建知识库
    # builder = KnowledgeBaseBuilder(doc_dir="./company_docs")
    # vectorstore = builder.build()

    # 后续运行：直接加载已有知识库
    qa_system = QASystem()

    # 模拟用户提问
    test_questions = [
        "公司的年假制度是怎样的？",
        "如何申请服务器资源？",
        "产品A的最新版本支持哪些数据库？",
        "刚才说的年假，试用期员工有吗？"
    ]

    for q in test_questions:
        print(f"\n{'='*60}")
        print(f"用户: {q}")
        print(f"{'='*60}")

        result = qa_system.chat(q)
        print(f"\n助手: {result['answer']}")
        print(f"\n[调试信息]")
        print(f"  召回文档数: {len(result['retrieved_sources'])}")
        print(f"  历史轮数: {result['history_length']}")
        for i, src in enumerate(result['retrieved_sources']):
            print(f"  来源{i+1}: {src['metadata']}")

7.5 运行效果示例

============================================================
用户: 公司的年假制度是怎样的？
============================================================

助手: 根据公司《员工手册》第3章规定：
1. 正式员工每年享有5-15天带薪年假，按工龄计算
2. 年假需提前3个工作日通过OA系统申请
3. 年假可拆分使用，但单次不少于0.5天
4. 未休年假可按日工资300%折算或顺延至次年3月

[调试信息]
  召回文档数: 5
  历史轮数: 2
  来源1: {'source': './docs/员工手册.pdf', 'page': 12}
  来源2: {'source': './docs/HR政策.txt', 'page': 0}

============================================================
用户: 刚才说的年假，试用期员工有吗？
============================================================

助手: 根据刚才提到的《员工手册》以及补充规定：
试用期员工不享有带薪年假。转正后按当年剩余月份比例折算年假天数。

[调试信息]
  召回文档数: 5
  历史轮数: 4
  来源1: {'source': './docs/员工手册.pdf', 'page': 15}

7.6 扩展方向

扩展功能	实现方式
权限控制	在文档 metadata 中标记权限等级，检索时过滤
多语言支持	使用多语言 Embedding 模型，或增加翻译中间层
答案溯源	在输出中标注答案来源文档及页码
反馈优化	收集用户点赞/点踩，用于微调检索排序
增量更新	监听文档目录变化，自动重新加载变更文档

8. 生产环境最佳实践

8.1 性能优化

# 1. 缓存机制
from langchain.globals import set_llm_cache
from langchain_community.cache import SQLiteCache
set_llm_cache(SQLiteCache(database_path="./llm_cache.db"))

# 2. 批处理嵌入
texts = ["文本1", "文本2", "文本3"]
embeddings_list = embeddings.embed_documents(texts)

# 3. 异步调用
import asyncio
result = await chain.ainvoke({"query": "问题"})

# 4. 流式输出
for chunk in chain.stream({"query": "问题"}):
    yield chunk.content

8.2 安全与监控

# 1. 敏感信息过滤
from langchain_core.runnables import RunnableLambda
import re

def filter_sensitive_info(text: str) -> str:
    text = re.sub(r'\d{17}[\dXx]', '[ID_CARD]', text)
    text = re.sub(r'1[3-9]\d{9}', '[PHONE]', text)
    return text

safe_chain = chain | RunnableLambda(filter_sensitive_info)

# 2. LangSmith 监控
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "production-app"

# 3. 错误重试
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def robust_invoke(chain, inputs):
    return chain.invoke(inputs)

8.3 部署方案

# 使用 LangServe 部署为 API
from langserve import add_routes
from fastapi import FastAPI

app = FastAPI(title="LangChain API", version="1.0")
add_routes(app, qa_system.qa_chain, path="/qa")

# 启动: uvicorn app:app --host 0.0.0.0 --port 8000

9. 总结与展望

9.1 LangChain 核心优势

生态最完善：97,000+ GitHub Star，50,000+ 生产应用，最多的第三方集成
灵活性最高：模块化设计，可深度定制业务流程
RAG 能力强：文档处理链路成熟，检索性能优异
生产就绪：LangSmith 提供全生命周期管理，LangGraph 支持复杂编排

9.2 技术选型建议

需求场景	推荐方案
快速原型验证	LangChain + LCEL
企业级 RAG 系统	LangChain + LangGraph + LangSmith
多 Agent 协作	LangGraph 编排 + LangChain 工具
简单问答应用	直接使用 LLM API 即可
极简开发需求	考虑 Dify 等低代码平台