© 2026 DREAMVFIA UNION

一、引言：大语言模型应用开发的范式转变

在人工智能发展的历程中，2022年末ChatGPT的横空出世标志着生成式AI时代的正式到来。然而，如何将强大的大语言模型（Large Language Model，LLM）能力转化为实际的应用程序，一直是开发者面临的核心挑战。早期的LLM应用开发往往采用简单的API调用方式，这种方式虽然直接，但难以构建复杂、可靠的应用程序。随着应用场景的深入，开发者逐渐意识到，仅靠单一的模型调用远远不够，需要一套完整的框架来处理上下文管理、工具调用、记忆机制、检索增强等复杂需求。

LangChain正是为解决这些问题而生的开源框架。它由Harrison Chase于2022年10月创立，经过两年多的发展，已经成为构建LLM应用的事实标准。LangChain的核心设计理念是“组合优于继承”，通过模块化的组件设计，让开发者能够像搭积木一样灵活地构建各种复杂的LLM应用。从简单的问答系统到复杂的企业级知识助手，LangChain提供了完整的工具链支持。

本文将带领读者从LangChain的基础概念出发，通过大量的代码示例，逐步深入到实战项目的开发。我们将涵盖LangChain的核心组件、LCEL表达式语言、检索增强生成（RAG）架构、智能体（Agent）开发等关键主题，并通过一个完整的企业知识助手项目，将理论知识转化为实际可运行的应用程序。无论你是刚刚接触LLM开发的新人，还是希望系统化提升技能的开发者，本文都将为你提供有价值的参考。

二、环境准备与快速入门

2.1 开发环境搭建

在开始LangChain开发之前，我们需要准备好Python开发环境。LangChain支持Python 3.8及以上版本，推荐使用Python 3.10或更高版本以获得最佳兼容性。首先，让我们创建一个独立的虚拟环境来隔离项目依赖：

# 创建虚拟环境
python -m venv langchain-env

# 激活虚拟环境
# Linux/Mac:
source langchain-env/bin/activate
# Windows:
langchain-env\Scripts\activate

接下来安装LangChain的核心包及相关依赖。为了保持项目的模块化，LangChain将功能拆分成了多个独立的包。基础安装只需langchain-core，而实际开发中我们通常需要安装更多的社区组件：

# 安装LangChain核心
pip install langchain-core

# 安装LangChain主包（包含社区组件）
pip install langchain

# 安装OpenAI集成（根据使用的模型选择对应的集成包）
pip install langchain-openai

# 安装文本嵌入模型
pip install langchain-community

# 安装常用的向量数据库客户端
pip install faiss-cpu  # FAISS向量库
pip install chromadb   # Chroma向量库

# 安装文档加载器支持
pip install pypdf      # PDF支持
pip install python-dotenv  # 环境变量管理

# 安装开发辅助工具
pip install langsmith  # LangChain开发调试工具

环境变量管理是LLM应用开发的重要环节。为了保护API密钥等敏感信息，我们不应该将它们直接硬编码在代码中，而是使用环境变量或专门的配置文件。推荐创建一个.env文件来管理这些配置：

# .env 文件内容
OPENAI_API_KEY=sk-your-api-key-here
ANTHROPIC_API_KEY=sk-ant-your-api-key-here
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=your-langsmith-key

# 在代码中加载环境变量
from dotenv import load_dotenv
import os

load_dotenv()  # 加载.env文件

# 验证环境变量加载
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
    raise ValueError("请在.env文件中设置OPENAI_API_KEY")
print(f"API密钥已加载，长度: {len(api_key)}字符")

2.2 第一个LangChain程序

让我们从最简单的例子开始，理解LangChain的基本工作流程。LangChain应用的核心流程可以概括为：接收输入、调用模型、处理输出。这个流程看似简单，但LangChain提供了丰富的抽象来简化这个过程：

from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage

# 初始化聊天模型
# 默认使用gpt-3.5-turbo模型
chat = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.7,
    max_tokens=1000
)

# 构建消息列表
messages = [
    SystemMessage(content="你是一位专业的Python技术作家，擅长用通俗易懂的语言解释技术概念。"),
    HumanMessage(content="请用一句话解释什么是装饰器")
]

# 调用模型获取响应
response = chat.invoke(messages)

# 输出结果
print(f"AI回复: {response.content}")
print(f"回复类型: {type(response)}")
print(f"额外信息: {response.response_metadata}")

运行上述代码，你应该能看到AI对装饰器的解释。这个例子展示了LangChain的最基本用法：创建模型实例、构建提示、获取响应。但在实际应用中，我们需要更强大的功能来处理复杂的业务需求。

三、LangChain核心组件详解

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

Model I/O是LangChain最核心的组件，它封装了与各种语言模型交互的逻辑。在深入了解之前，我们需要区分两个重要概念：LLMs和Chat Models。传统的LLMs接受文本输入并输出文本，而Chat Models则接受消息列表（包含角色信息）并输出消息。在现代的GPT API中，实际调用的都是Chat Models，但LangChain保留了两种接口以保持兼容性：

from langchain_openai import OpenAI, ChatOpenAI
from langchain.schema import HumanMessage, AIMessage, SystemMessage

# 传统LLM接口（逐渐被废弃）
llm = OpenAI(model="gpt-3.5-turbo-instruct")
llm_response = llm.invoke("解释什么是量子计算")
print(f"LLM输出: {llm_response}")

# Chat Model接口（推荐）
chat = ChatOpenAI(model="gpt-3.5-turbo")
chat_response = chat.invoke([
    HumanMessage(content="解释什么是量子计算")
])
print(f"Chat输出: {chat_response.content}")

提示模板（Prompt Templates）是提高代码复用性的重要工具。在实际应用中，我们通常需要根据不同的用户输入动态构建提示。通过提示模板，我们可以定义提示的结构，只替换其中的变量部分：

from langchain.prompts import PromptTemplate, ChatPromptTemplate, HumanMessagePromptTemplate

# 简单的文本提示模板
simple_template = PromptTemplate.from_template(
    "请用{language}语言解释{concept}概念，限制在{length}个字以内。"
)

# 生成具体的提示
prompt = simple_template.format(
    language="中文",
    concept="闭包",
    length="100"
)
print(f"生成的提示: {prompt}")

# 聊天提示模板（用于Chat Models）
chat_template = ChatPromptTemplate.from_messages([
    ("system", "你是一位{domain}领域的专家。"),
    ("human", "我想了解{topic}，请用通俗易懂的方式解释。"),
    ("ai", "好的，我会用适合初学者的方式解释{topic}。"),
    ("human", "那么，{question}？")
])

# 填充模板
chat_prompt = chat_template.format_messages(
    domain="计算机科学",
    topic="算法",
    question="排序算法的时间复杂度"
)

for msg in chat_prompt:
    print(f"角色: {msg.type}, 内容: {msg.content}")

few-shot学习是提升模型表现的有效技术。通过在提示中提供少量示例，可以显著提高模型在特定任务上的准确性。LangChain提供了专门的支持：

from langchain.prompts import FewShotPromptTemplate, PromptTemplate

# 定义示例
examples = [
    {"input": "2+2", "output": "4"},
    {"input": "5+3", "output": "8"},
    {"input": "10-4", "output": "6"},
]

# 创建示例格式化器
example_prompt = PromptTemplate(
    input_variables=["input", "output"],
    template="问题: {input}\n答案: {output}"
)

# 创建few-shot提示模板
few_shot_template = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    suffix="问题: {input}\n答案:",
    input_variables=["input"]
)

# 使用示例
prompt = few_shot_template.format(input="7+8")
print(prompt)

输出解析器（Output Parsers）是另一个重要的组件。当我们需要模型返回结构化数据（如JSON、XML）时，输出解析器可以帮助我们将模型的原始输出转换为Python对象：

from langchain.output_parsers import PydanticOutputParser
from langchain.pydantic_v1 import BaseModel
from typing import List

# 定义期望的输出结构
class Recipe(BaseModel):
    name: str
    ingredients: List[str]
    steps: List[str]
    cooking_time: int  # 分钟

# 创建解析器
parser = PydanticOutputParser(pydantic_object=Recipe)

# 创建提示模板
prompt = PromptTemplate(
    template="请生成一个简单的{ cuisine }菜谱。{format_instructions}",
    input_variables=["cuisine"],
    partial_variables={"format_instructions": parser.get_format_instructions()}
)

# 调用模型
from langchain_openai import ChatOpenAI
chain = prompt | chat | parser

# 执行
result = chain.invoke({"cuisine": "意大利"})
print(f"菜谱名称: {result.name}")
print(f"食材: {result.ingredients}")
print(f"步骤: {result.steps}")
print(f"烹饪时间: {result.cooking_time}分钟")

3.2 数据连接与RAG基础

检索增强生成（Retrieval-Augmented Generation，RAG）是当前LLM应用最流行的架构之一。RAG的核心思想是让模型在生成回答时能够访问外部知识库，从而减少幻觉、提高回答的准确性。LangChain提供了完整的数据处理工具链来支持RAG应用开发。

文档加载器（Document Loaders）是RAG pipeline的起点。LangChain支持从各种数据源加载文档，包括PDF、HTML、Markdown、数据库等：

# 文本文件加载
from langchain_community.document_loaders import TextLoader, PyPDFLoader, DirectoryLoader
from langchain_community.document_loaders import UnstructuredMarkdownLoader

# 加载文本文件
loader = TextLoader("example.txt")
documents = loader.load()

# 加载PDF文件
pdf_loader = PyPDFLoader("example.pdf")
pdf_documents = pdf_loader.load()

# 加载整个目录下的文档
dir_loader = DirectoryLoader(
    "./documents",
    glob="**/*.txt",
    loader_cls=TextLoader
)
all_docs = dir_loader.load()

# 查看加载的文档
for doc in documents:
    print(f"内容长度: {len(doc.page_content)}")
    print(f"元数据: {doc.metadata}")

文档转换器（Document Transformers）用于将加载的原始文档切分成更小的块，以便更好地被嵌入模型处理和检索。文本切分看似简单，实际上有很多需要注意的细节：

from langchain.text_splitter import RecursiveCharacterTextSplitter, CharacterTextSplitter

# 递归字符文本分割器（推荐）
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # 每个块的最大字符数
    chunk_overlap=200,    # 块之间的重叠字符数
    length_function=len, # 用于计算长度的函数
    separators=["\n\n", "\n", " ", ""]  # 分隔符优先级
)

# 原始文档
raw_text = """
这是第一段内容。这里是更多的内容。
这是第二段内容，包含了一些重要的信息。

这是第三段，是一个新的章节开始。
"""

# 执行分割
docs = text_splitter.create_documents([raw_text])

for i, doc in enumerate(docs):
    print(f"--- 文档块 {i+1} ---")
    print(doc.page_content)
    print(f"元数据: {doc.metadata}")

嵌入模型（Embedding Models）将文本转换为向量表示，是实现语义检索的关键。LangChain提供了统一的接口来使用各种嵌入模型：

from langchain_community.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings

# OpenAI嵌入（需要API密钥）
openai_embeddings = OpenAIEmbeddings(
    model="text-embedding-ada-002"
)

# HuggingFace本地嵌入模型（免费）
hf_embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

# 生成嵌入向量
texts = [
    "今天天气真好",
    "天气很棒",
    "python是一门编程语言"
]

# 使用OpenAI嵌入
embeddings = openai_embeddings.embed_documents(texts)
print(f"嵌入向量维度: {len(embeddings[0])}")

from numpy import dot
from numpy.linalg import norm

def cosine_similarity(a, b):
    return dot(a, b) / (norm(a) * norm(b))

print(f"文本1与文本2的相似度: {cosine_similarity([embeddings[0]], [embeddings[1]])[0][0]:.4f}")
print(f"文本1与文本3的相似度: {cosine_similarity([embeddings[0]], [embeddings[2]])[0][0]:.4f}")

向量存储（Vector Stores）用于存储和检索嵌入向量。LangChain支持多种向量数据库，包括Chroma、FAISS、Pinecone、Weaviate等：

from langchain_community.vectorstores import Chroma, FAISS
from langchain_community.docstore.in_memory import InMemoryDocstore

# 使用Chroma向量数据库
texts = [
    "LangChain是一个用于构建LLM应用的框架",
    "RAG是检索增强生成技术",
    "向量数据库用于存储嵌入向量"
]

# 创建Chroma向量库
vectorstore = Chroma.from_texts(
    texts=texts,
    embedding=openai_embeddings,
    collection_name="my_collection"
)

# 执行相似性检索
query = "什么是LangChain？"
docs = vectorstore.similarity_search(query, k=2)

print(f"查询: {query}")
print(f"检索结果:")
for i, doc in enumerate(docs):
    print(f"  {i+1}. {doc.page_content}")

3.3 Chain与LCEL表达式语言

LangChain Expression Language（LCEL）是LangChain最重要的创新之一。它提供了一种声明式的方式来组合各种组件，创建复杂的处理管道。LCEL使用管道操作符|，将上一个组件的输出传递给下一个组件的输入：

from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.schema import StrOutputParser

# 基础LCEL链
llm = ChatOpenAI(model="gpt-3.5-turbo")

prompt = PromptTemplate.from_template(
    "用一句话解释{concept}："
)

# 使用 | 运算符组合组件
chain = prompt | llm | StrOutputParser()

# 执行链
result = chain.invoke({"concept": "量子纠缠"})
print(result)

LCEL的真正强大之处在于它支持复杂的链式组合，包括并行处理、条件分支、错误处理等：

from langchain.schema import RunnableParallel, RunnableBranch

# 创建处理分支
positive_template = PromptTemplate.from_template(
    "用户表达了积极情绪：{input}。请用鼓励的语气回应。"
)
negative_template = PromptTemplate.from_template(
    "用户表达了消极情绪：{input}。请用安慰的语气回应。"
)
neutral_template = PromptTemplate.from_template(
    "用户表达了中性情绪：{input}。请用专业的语气回应。"
)

# 定义分支逻辑
def classify_emotion(text):
    positive_words = ["好", "棒", "喜欢", "开心", "优秀"]
    negative_words = ["差", "糟", "讨厌", "难过", "失望"]
    
    if any(word in text for word in positive_words):
        return "positive"
    elif any(word in text for word in negative_words):
        return "negative"
    else:
        return "neutral"

# 使用RunnableBranch实现条件分支
branch = RunnableBranch(
    (lambda x: classify_emotion(x["input"]) == "positive", positive_template | llm | StrOutputParser()),
    (lambda x: classify_emotion(x["input"]) == "negative", negative_template | llm | StrOutputParser()),
    neutral_template | llm | StrOutputParser()
)

# 主链
main_chain = (
    {"input": lambda x: x["input"]}
    | branch
)

# 测试
test_inputs = [
    "今天天气真好，心情很开心！",
    "这个产品太糟糕了，很失望。",
    "请介绍一下你们的服务的特点。"
]

for test_input in test_inputs:
    result = main_chain.invoke({"input": test_input})
    print(f"输入: {test_input}")
    print(f"输出: {result}")
    print("-" * 50)

四、Memory与状态管理

4.1 对话内存基础

在构建聊天机器人或对话系统时，保持上下文一致性是基本要求。LangChain的Memory组件提供了多种内存实现来处理对话历史：

from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI

# 初始化模型
llm = ChatOpenAI(model="gpt-3.5-turbo")

# 方法1：缓冲内存（保留完整对话）
buffer_memory = ConversationBufferMemory(
    return_messages=True,  # 返回消息对象而非字符串
    output_key="response",  # 指定输出键名
    input_key="input"        # 指定输入键名
)

# 方法2：摘要内存（压缩历史信息，适合长对话）
summary_memory = ConversationSummaryMemory(
    llm=llm,
    return_messages=True
)

# 创建对话链
conversation = ConversationChain(
    llm=llm,
    memory=buffer_memory,
    verbose=True
)

# 进行多轮对话
response1 = conversation.predict(input="你好！我叫张三。")
print(f"AI: {response1}")

response2 = conversation.predict(input="我刚才告诉你我的名字是什么？")
print(f"AI: {response2}")

response3 = conversation.predict(input="我喜欢机器学习和人工智能。")
print(f"AI: {response3}")

# 查看内存中存储的内容
print("\n=== 对话历史 ===")
print(buffer_memory.chat_memory.messages)

4.2 自定义内存实现

对于复杂的应用场景，我们可能需要实现自定义的内存策略。以下是一个基于滑动窗口的自定义内存实现：

from langchain.memory import BaseMemory
from langchain.schema import HumanMessage, AIMessage
from typing import List, Dict, Any

class SlidingWindowMemory(BaseMemory):
    """滑动窗口内存：只保留最近N轮对话"""
    
    def __init__(self, k: int = 5):
        self.k = k
        self.messages: List[Dict[str, str]] = []
    
    @property
    def memory_variables(self) -> List[str]:
        return ["history"]
    
    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        """加载内存变量"""
        history_str = "\n".join([
            f"用户: {msg['content']}" if msg['type'] == 'human' else f"AI: {msg['content']}"
            for msg in self.messages
        ])
        return {"history": history_str}
    
    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, Any]) -> None:
        """保存对话上下文"""
        input_text = inputs.get("input", "")
        output_text = outputs.get("response", outputs.get("output", ""))
        
        self.messages.append({"type": "human", "content": input_text})
        self.messages.append({"type": "ai", "content": output_text})
        
        if len(self.messages) > self.k * 2:
            self.messages = self.messages[-self.k * 2:]
    
    def clear(self) -> None:
        """清空内存"""
        self.messages = []

# 使用自定义内存
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

custom_memory = SlidingWindowMemory(k=3)

prompt = PromptTemplate.from_template("""
你是AI助手。请根据对话历史回复用户。

对话历史:
{history}

用户: {input}
AI:
""")

chain = LLMChain(llm=llm, prompt=prompt, memory=custom_memory)

print(chain.predict(input="你好！"))
print(chain.predict(input="今天天气不错。"))
print(chain.predict(input="你喜欢晴天还是雨天？"))
print("\n内存状态:")
print(custom_memory.load_memory_variables({}))

五、Agent与工具使用

5.1 Agent核心概念

Agent是LangChain最强大的功能之一，它让LLM能够自主决定使用哪些工具来完成复杂任务。Agent的核心工作流程可以概括为：接收用户输入、选择并执行工具、获取结果、决定是否继续、生成最终响应：

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.tools import Tool

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

# 定义工具
def calculate(expression: str) -> str:
    """执行数学计算"""
    try:
        result = eval(expression)
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")

# 创建工具列表
tools = [
    Tool(
        name="计算器",
        func=calculate,
        description="用于数学计算。输入应该是数学表达式，如'2+3*5'"
    ),
    Tool(
        name="时间查询",
        func=get_current_time,
        description="获取当前日期和时间"
    )
]

# 创建提示模板
prompt = PromptTemplate.from_template("""
你是一个AI助手，可以调用工具来完成任务。

可用工具:
- 计算器: 用于数学计算
- 时间查询: 获取当前时间

对话历史:
{chat_history}

用户输入: {input}
{agent_scratchpad}

请按照以下格式思考和行动：
1. 思考是否需要使用工具
2. 如果需要，选择合适的工具并给出输入
3. 观察工具返回的结果
4. 最终给出回答

注意：只有需要精确计算或获取实时信息时才使用工具。
""")

# 创建Agent
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 执行Agent
result = agent_executor.invoke({
    "input": "请计算123乘以456，然后告诉我现在的时间。",
    "chat_history": ""
})
print(f"\n最终结果: {result['output']}")

5.2 自定义工具开发

在实际应用中，我们经常需要创建自定义工具来满足特定的业务需求。以下是一个完整的自定义工具开发示例：

from langchain.tools import Tool, StructuredTool
from langchain_core.utils.function_calling import convert_to_openai_function
import requests

# 方式1：简单函数工具
def search_wikipedia(query: str) -> str:
    """搜索维基百科"""
    url = f"https://zh.wikipedia.org/w/api.php"
    params = {
        "action": "query",
        "list": "search",
        "srsearch": query,
        "format": "json"
    }
    response = requests.get(url, params=params)
    data = response.json()
    
    if data["query"]["search"]:
        result = data["query"]["search"][0]
        return f"{result['title']}: {result['snippet']}..."
    return "未找到相关结果"

# 方式2：带结构化输出的工具
def get_weather(location: str, unit: str = "celsius") -> dict:
    """获取天气信息（模拟）"""
    return {
        "location": location,
        "temperature": 22 if unit == "celsius" else 71.6,
        "unit": unit,
        "condition": "晴",
        "humidity": 65
    }

# 创建工具
wiki_tool = Tool(
    name="维基百科搜索",
    func=search_wikipedia,
    description="当用户询问事实性信息或需要了解某个概念时使用"
)

# 使用StructuredTool创建复杂工具
weather_tool = StructuredTool.from_function(
    func=get_weather,
    name="天气查询",
    description="查询指定位置的天气信息，需要提供location参数"
)

# 创建工具集
from langchain.tools import ToolKit

toolkit = ToolKit(
    tools=[wiki_tool, weather_tool]
)

# Agent中使用工具
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.prompts import PromptTemplate

prompt = PromptTemplate.from_template("""
你是一个知识丰富的AI助手。

用户问题: {input}

你可以使用以下工具:
- 维基百科搜索: 搜索事实性信息
- 天气查询: 查询天气（参数: location, unit）

如果问题需要使用工具，请调用相应的工具。
不要捏造信息。
""")

agent = create_openai_functions_agent(llm, toolkit.get_tools(), prompt)
executor = AgentExecutor(agent=agent, tools=toolkit.get_tools(), verbose=True)

# 测试
result1 = executor.invoke({"input": "北京现在的天气怎么样？"})
print(f"天气查询结果: {result1['output']}\n")

result2 = executor.invoke({"input": "请介绍一下量子计算的基本原理。"})
print(f"维基百科搜索结果: {result2['output']}")

5.3 ReAct框架与Agent进阶

ReAct（Reasoning + Acting）是一种让Agent同时具备推理和行动能力的框架。LangChain内置了对ReAct的支持，可以构建更智能的Agent：

from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub

# 从Hub获取ReAct提示模板
react_prompt = hub.pull("hwchase17/react")

# 打印提示内容
print("ReAct提示模板:")
print(react_prompt.template)

# 使用自定义工具创建ReAct Agent
def search_code_documentation(query: str) -> str:
    """搜索编程文档"""
    docs = {
        "python": "Python是一种高级编程语言，强调代码可读性。",
        "javascript": "JavaScript是一种脚本语言，用于Web开发。",
        "java": "Java是一种面向对象的编程语言，'一次编写，到处运行'。",
        "rust": "Rust是一种系统编程语言，强调内存安全。",
    }
    
    query_lower = query.lower()
    for key, value in docs.items():
        if key in query_lower:
            return value
    return "未找到相关文档"

code_tool = Tool(
    name="代码文档搜索",
    func=search_code_documentation,
    description="搜索编程语言的文档信息"
)

# 创建ReAct Agent
react_agent = create_react_agent(llm, [code_tool], react_prompt)
react_executor = AgentExecutor(
    agent=react_agent, 
    tools=[code_tool],
    verbose=True,
    handle_parsing_errors=True
)

# 测试ReAct Agent
result = react_executor.invoke({
    "input": "Python和Java有什么区别？"
})
print(f"\n最终回答: {result['output']}")

六、实战项目：企业知识助手

6.1 项目架构设计

现在我们将综合运用LangChain的各种组件，构建一个完整的企业知识助手系统。该系统将具备以下功能：文档上传与处理、向量知识库构建、智能问答、来源引用返回：

# 项目完整代码 - 企业知识助手

import os
from dotenv import load_dotenv
load_dotenv()

from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder
from langchain.chains import create_retrieval_chain, create_history_aware_retriever
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.memory import ConversationBufferMemory
from typing import List

# ==================== 配置类定义 ====================
class Config:
    """应用配置"""
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    MODEL_NAME = "gpt-3.5-turbo"
    EMBEDDING_MODEL = "text-embedding-ada-002"
    PERSIST_DIRECTORY = "./vectorstore"
    
config = Config()

# ==================== 文档处理模块 ====================
class DocumentProcessor:
    """文档处理器"""
    
    def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", "。", "！", "？", " ", ""]
        )
    
    def load_pdf(self, file_path: str):
        """加载PDF文档"""
        loader = PyPDFLoader(file_path)
        return loader.load()
    
    def load_text(self, file_path: str):
        """加载文本文档"""
        loader = TextLoader(file_path, encoding="utf-8")
        return loader.load()
    
    def process_documents(self, documents: List):
        """处理文档：分割"""
        return self.text_splitter.split_documents(documents)

# ==================== 知识库管理 ====================
class KnowledgeBase:
    """知识库管理"""
    
    def __init__(self, embedding_model=None):
        self.embedding_model = embedding_model or OpenAIEmbeddings(
            model=config.EMBEDDING_MODEL
        )
        self.vectorstore = None
    
    def create_from_documents(self, documents, collection_name: str = "knowledge_base"):
        """从文档创建向量知识库"""
        self.vectorstore = Chroma.from_documents(
            documents=documents,
            embedding=self.embedding_model,
            collection_name=collection_name
        )
        return self.vectorstore
    
    def load_existing(self, collection_name: str = "knowledge_base"):
        """加载已存在的知识库"""
        self.vectorstore = Chroma(
            client=None,
            collection_name=collection_name,
            embedding_function=self.embedding_model
        )
        return self.vectorstore
    
    def get_retriever(self, search_kwargs: dict = None):
        """获取检索器"""
        if not self.vectorstore:
            raise ValueError("知识库未初始化")
        
        default_kwargs = {"k": 3}
        if search_kwargs:
            default_kwargs.update(search_kwargs)
        
        return self.vectorstore.as_retriever(
            search_type="similarity",
            search_kwargs=default_kwargs
        )

# ==================== RAG链构建 ====================
class RAGChainBuilder:
    """RAG链构建器"""
    
    def __init__(self, llm, retriever):
        self.llm = llm
        self.retriever = retriever
    
    def build_qa_chain(self):
        """构建问答链"""
        system_prompt = """你是一个企业知识助手，负责根据提供的文档内容回答用户的问题。

要求：
1. 只根据提供的上下文信息回答，不要编造信息
2. 如果上下文中没有相关信息，请明确告知用户
3. 回答要简洁明了，引用相关来源
4. 使用中文回答用户的问题

上下文信息：
{context}
"""
        
        question_answer_chain = create_stuff_documents_chain(
            self.llm,
            PromptTemplate.from_template(system_prompt)
        )
        
        rag_chain = create_retrieval_chain(
            self.retriever,
            question_answer_chain
        )
        
        return rag_chain
    
    def build_conversational_chain(self):
        """构建带上下文的对话链"""
        contextualize_q_prompt = PromptTemplate.from_template(
            """给定聊天历史和最新用户问题，
            这个问题可能涉及之前对话中的上下文。
            请将当前问题改写为一个独立的问题，使其不依赖聊天历史也能被理解。

            聊天历史：
            {chat_history}

            当前问题：{input}

            改写后的问题："""
        )
        
        history_aware_retriever = create_history_aware_retriever(
            self.llm,
            self.retriever,
            contextualize_q_prompt
        )
        
        system_prompt = """你是一个专业的企业知识助手。基于提供的文档内容和聊天历史来回答用户的问题。

如果文档中的信息不足以回答问题，请如实告知用户。
"""
        
        qa_chain = create_stuff_documents_chain(
            self.llm,
            ChatPromptTemplate.from_messages([
                ("system", system_prompt),
                MessagesPlaceholder(variable_name="chat_history"),
                ("human", "{input}")
            ])
        )
        
        conversational_rag = create_retrieval_chain(
            history_aware_retriever,
            qa_chain
        )
        
        return conversational_rag

# ==================== 主应用类 ====================
class EnterpriseKnowledgeAssistant:
    """企业知识助手主类"""
    
    def __init__(self):
        self.llm = ChatOpenAI(
            model=config.MODEL_NAME,
            temperature=0.3
        )
        
        self.document_processor = DocumentProcessor()
        self.knowledge_base = KnowledgeBase()
        self.memory = ConversationBufferMemory(
            return_messages=True,
            output_key="answer",
            input_key="input"
        )
        
        self.qa_chain = None
        self.conversational_chain = None
    
    def ingest_document(self, file_path: str):
        """摄取文档到知识库"""
        if file_path.endswith('.pdf'):
            documents = self.document_processor.load_pdf(file_path)
        elif file_path.endswith('.txt'):
            documents = self.document_processor.load_text(file_path)
        else:
            raise ValueError(f"不支持的文件类型: {file_path}")
        
        processed_docs = self.document_processor.process_documents(documents)
        print(f"文档已分割为 {len(processed_docs)} 个块")
        
        self.knowledge_base.create_from_documents(processed_docs)
        print("知识库创建完成")
        
        retriever = self.knowledge_base.get_retriever()
        chain_builder = RAGChainBuilder(self.llm, retriever)
        self.qa_chain = chain_builder.build_qa_chain()
        self.conversational_chain = chain_builder.build_conversational_chain()
        
        return len(processed_docs)
    
    def ask(self, question: str, return_sources: bool = True):
        """提问"""
        if not self.qa_chain:
            raise ValueError("请先摄取文档")
        
        result = self.qa_chain.invoke({"input": question})
        
        answer = result["answer"]
        
        if return_sources:
            sources = []
            for doc in result["context"]:
                sources.append({
                    "content": doc.page_content[:200] + "...",
                    "metadata": doc.metadata
                })
            return {
                "answer": answer,
                "sources": sources
            }
        
        return {"answer": answer}
    
    def conversational_ask(self, question: str):
        """对话式提问（带记忆）"""
        if not self.conversational_chain:
            raise ValueError("请先摄取文档")
        
        chat_history = self.memory.load_memory_variables({}).get("history", "")
        
        result = self.conversational_chain.invoke({
            "input": question,
            "chat_history": chat_history
        })
        
        self.memory.save_context(
            {"input": question},
            {"answer": result["answer"]}
        )
        
        return result["answer"]

# ==================== 使用示例 ====================
def main():
    """主函数"""
    
    assistant = EnterpriseKnowledgeAssistant()
    
    test_content = """
    公司简介
    
    DREAMVFIA UNION是一家专注于人工智能技术研发的公司。
    成立于2023年，总部位于北京。
    
    核心技术
    
    公司主要技术方向包括：
    1. 大语言模型应用开发
    2. 量子计算技术
    3. 机器学习平台
    
    产品服务
    
    主要产品包括：
    - AI开发工具链
    - 企业级知识管理系统
    - 智能对话平台
    
    联系方式
    
    邮箱: contact@dreamvfia.com
    电话: 400-888-8888
    地址: 北京市海淀区
    """
    
    with open("company_info.txt", "w", encoding="utf-8") as f:
        f.write(test_content)
    
    print("开始摄取文档...")
    assistant.ingest_document("company_info.txt")
    
    print("\n" + "="*50)
    print("知识助手已准备就绪！")
    print("="*50 + "\n")
    
    questions = [
        "DREAMVFIA UNION的核心技术是什么？",
        "公司的产品有哪些？",
        "公司的联系方式是什么？"
    ]
    
    for question in questions:
        print(f"\n问题: {question}")
        result = assistant.ask(question)
        print(f"回答: {result['answer']}")
        print(f"来源数量: {len(result['sources'])}")

if __name__ == "__main__":
    main()

6.2 Streamlit界面开发

为了让知识助手更易于使用，我们可以使用Streamlit快速构建Web界面：

# app.py - Streamlit Web界面

import streamlit as st
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OpenAIEmbeddings
from langchain.prompts import PromptTemplate
from langchain.chains import create_retrieval_chain, create_stuff_documents_chain
from langchain.memory import ConversationBufferMemory
import os
from dotenv import load_dotenv

load_dotenv()

# 页面配置
st.set_page_config(
    page_title="企业知识助手",
    page_icon="🤖",
    layout="wide"
)

# 初始化会话状态
if "messages" not in st.session_state:
    st.session_state.messages = []

if "vectorstore" not in st.session_state:
    st.session_state.vectorstore = None

if "memory" not in st.session_state:
    st.session_state.memory = ConversationBufferMemory(
        return_messages=True,
        output_key="answer",
        input_key="input"
    )

# 侧边栏
with st.sidebar:
    st.title("📚 知识库管理")
    
    uploaded_file = st.file_uploader(
        "上传文档",
        type=["txt", "pdf"],
        help="支持TXT和PDF格式"
    )
    
    if uploaded_file:
        if st.button("处理文档"):
            with st.spinner("正在处理文档..."):
                import tempfile
                from langchain_community.document_loaders import PyPDFLoader, TextLoader
                from langchain.text_splitter import RecursiveCharacterTextSplitter
                
                with tempfile.NamedTemporaryFile(delete=False, suffix=uploaded_file.name) as tmp_file:
                    tmp_file.write(uploaded_file.getvalue())
                    tmp_path = tmp_file.name
                
                if uploaded_file.name.endswith('.pdf'):
                    loader = PyPDFLoader(tmp_path)
                else:
                    loader = TextLoader(tmp_path, encoding='utf-8')
                
                documents = loader.load()
                text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
                docs = text_splitter.split_documents(documents)
                
                embeddings = OpenAIEmbeddings()
                st.session_state.vectorstore = Chroma.from_documents(
                    docs, 
                    embeddings,
                    collection_name="knowledge_base"
                )
                
                st.success(f"文档处理完成！共 {len(docs)} 个知识块")
    
    if st.button("清空对话历史"):
        st.session_state.messages = []
        st.session_state.memory.clear()
        st.rerun()

# 主界面
st.title("🤖 企业知识助手")

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

if prompt := st.chat_input("请输入您的问题..."):
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)
    
    with st.chat_message("assistant"):
        with st.spinner("思考中..."):
            if st.session_state.vectorstore is None:
                response = "请先在侧边栏上传文档！"
            else:
                llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)
                retriever = st.session_state.vectorstore.as_retriever(k=3)
                
                system_prompt = """你是一个企业知识助手。根据提供的文档内容回答用户的问题。
                如果文档中没有相关信息，请如实告知用户。"""
                
                qa_chain = create_stuff_documents_chain(
                    llm,
                    PromptTemplate.from_template(system_prompt)
                )
                
                rag_chain = create_retrieval_chain(retriever, qa_chain)
                result = rag_chain.invoke({"input": prompt})
                response = result["answer"]
                
                st.session_state.memory.save_context(
                    {"input": prompt},
                    {"answer": response}
                )
            
            st.markdown(response)
            st.session_state.messages.append({"role": "assistant", "content": response})

运行命令：

pip install streamlit
streamlit run app.py

七、性能优化与生产部署

7.1 缓存策略

在生产环境中，减少API调用是降低成本的关键。LangChain提供了多种缓存策略：

from langchain.cache import InMemoryCache, SQLiteCache
from langchain.globals import set_llm_cache

# 内存缓存
set_llm_cache(InMemoryCache())

# SQLite缓存（持久化）
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

# 自定义缓存示例
class SemanticCache:
    """语义缓存：基于嵌入向量的相似度匹配"""
    
    def __init__(self, threshold: float = 0.95):
        self.cache = {}
        self.threshold = threshold
        from langchain_community.embeddings import OpenAIEmbeddings
        self.embeddings = OpenAIEmbeddings()
    
    def _get_cache_key(self, prompt: str) -> str:
        import hashlib
        return hashlib.md5(prompt.encode()).hexdigest()
    
    def get(self, prompt: str) -> str:
        key = self._get_cache_key(prompt)
        return self.cache.get(key)
    
    def set(self, prompt: str, response: str):
        key = self._get_cache_key(prompt)
        self.cache[key] = response
    
    def clear(self):
        self.cache = {}

semantic_cache = SemanticCache(threshold=0.95)

def cached_llm_call(prompt: str, llm):
    cached_response = semantic_cache.get(prompt)
    if cached_response:
        print(f"命中缓存: {prompt[:30]}...")
        return cached_response
    
    response = llm.invoke(prompt)
    response_text = response.content if hasattr(response, 'content') else str(response)
    semantic_cache.set(prompt, response_text)
    return response_text

7.2 LangSmith调试与监控

LangSmith是LangChain官方提供的调试和监控工具：

# 启用LangSmith追踪
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "my-langchain-app"

# 之后的每次LLM调用都会被LangSmith追踪
# 访问 https://smith.langchain.com 查看追踪结果

7.3 Docker部署

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

RUN apt-get update && apt-get install -y gcc && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

ENV PYTHONUNBUFFERED=1

EXPOSE 8501

CMD ["streamlit", "run", "app.py", "--server.address", "0.0.0.0"]

八、总结与展望

本文系统地介绍了LangChain框架的核心概念和实战技巧。从环境搭建开始，我们学习了如何配置开发环境、管理API密钥；深入探讨了Model I/O组件，掌握了提示模板和输出解析器的使用；详细讲解了数据连接模块，包括文档加载、文本分割、嵌入向量生成和向量数据库操作；通过LCEL表达式语言，实现了复杂链式组合；探索了Memory机制和Agent智能体，赋予了应用对话记忆和自主工具使用能力；最后，通过完整的企业知识助手项目，将所有知识点串联起来，形成了一个可实际运行的应用程序。

LangChain作为LLM应用开发的基础设施，正在快速迭代发展。随着LangGraph的推出，LangChain生态系统变得更加完善，为构建复杂的多Agent系统提供了更强的支持。对于想要深入LLM应用开发的读者，建议继续关注以下方向：LangGraph的深入学习、Agent工作流的优化、生产环境的监控与安全、以及大模型成本优化策略。

---

版权声明：© 2026 DREAMVFIA UNION