LangChain 实战：构建企业级 AI Agent 应用

前言

LangChain 是目前最流行的 LLM 应用开发框架，它提供了一套完整的工具链，帮助开发者快速构建基于大语言模型的智能应用。本文将带你从入门到精通。

一、LangChain 核心概念

1.1 LLM 与 Chains

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

# 基础 LLM 调用
llm = ChatOpenAI(model="gpt-4", temperature=0.7)

# Prompt 模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的{role}，擅长{skill}。"),
    ("user", "{question}")
])

# 构建简单的 Chain
chain = prompt | llm | StrOutputParser()

# 执行
result = chain.invoke({
    "role": "技术导师",
    "skill": "解释复杂概念",
    "question": "请解释什么是 RAG 技术？"
})

1.2 核心组件架构

┌─────────────────────────────────────────────┐
│                 LangChain                    │
├─────────────────────────────────────────────┤
│  Models    │    Chains    │    Agents       │
│  ├─ LLMs   │  ├─ Sequential│  ├─ ReAct       │
│  ├─ Chat   │  ├─ Router    │  ├─ Plan-and-Execute│
│  └─ Embed  │  └─ Parallel  │  └─ Self-Ask    │
├─────────────────────────────────────────────┤
│        Memory  │  Prompts  │  Output        │
│  ├─ Conversation│ ├─ Templates │ ├─ Parser   │
│  ├─ Entity      │ └─ Few-shot  │ └─ Tools    │
│  └─ Summary     │              │             │
└─────────────────────────────────────────────┘

二、实现企业级 RAG 系统

2.1 数据加载与处理

from langchain_community.document_loaders import DirectoryLoader, PDFPlumberLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 加载文档
loader = DirectoryLoader(
    "./data/",
    glob="**/*.pdf",
    loader_cls=PDFPlumberLoader
)
documents = loader.load()

# 智能分块
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
    separators=["\n\n", "\n", "。", ".", " ", ""]
)
chunks = text_splitter.split_documents(documents)

2.2 向量化存储

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.retrievers.multi_query import MultiQueryRetriever

# 创建嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

# 构建向量数据库
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 多查询检索器（提升召回率）
retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    llm=llm
)

2.3 完整的 RAG Pipeline

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

# RAG Prompt 模板
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个智能助手，基于提供的上下文回答问题。

上下文信息：
{context}

规则：
- 只基于提供的上下文回答
- 如果上下文中没有答案，明确告知
- 保持简洁准确
- 引用具体来源"""),
    ("user", "问题：{input}")
])

# 构建检索增强链
qa_chain = create_retrieval_chain(
    retriever,
    create_stuff_documents_chain(llm, rag_prompt)
)

# 执行查询
result = qa_chain.invoke({"input": "什么是 LangChain 的核心特性？"})
print(result["answer"])

三、构建智能 Agent

3.1 Agent 工具设计

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain.tools import tool

# 自定义工具
@tool
def search_knowledge_base(query: str) -> str:
    """搜索知识库中的相关信息"""
    docs = retriever.invoke(query)
    return "\n\n".join([doc.page_content for doc in docs])

@tool
def calculate_expression(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果: {result}"
    except:
        return "表达式有误"

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

tools = [search_knowledge_base, calculate_expression, get_current_time]

3.2 ReAct Agent

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent

# 创建 Agent
agent_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手，使用可用工具回答问题。"),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, agent_prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,
    early_stopping_method="generate"
)

# 运行 Agent
response = agent_executor.invoke({
    "input": "帮我查询公司Q1财报，然后计算增长率"
})

四、Memory 管理

4.1 对话历史

from langchain_core.messages import HumanMessage, AIMessage
from langchain_community.chat_message_histories import ChatMessageHistory

# 会话历史
history = ChatMessageHistory()

# 添加消息
history.add_user_message("你好")
history.add_ai_message("你好！有什么可以帮助你的？")

# 获取历史
messages = history.messages

4.2 对话链

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

memory = ConversationBufferMemory(
    return_messages=True,
    memory_key="chat_history"
)

conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 持续对话
conversation.invoke("介绍一下 Transformer")
conversation.invoke("它的多头注意力机制是怎么工作的？")

五、最佳实践

5.1 性能优化

• 使用 asyncio 实现并发调用
• 合理设置 chunk_size 和 overlap
• 启用向量数据库的缓存机制
• 使用连接池管理数据库连接

5.2 错误处理

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def safe_llm_call(prompt: str):
    try:
        response = await llm.ainvoke(prompt)
        return response
    except Exception as e:
        logger.error(f"LLM调用失败: {e}")
        raise

5.3 监控与日志

from langchain.callbacks import StdOutCallbackHandler

# 添加 callback 追踪
llm_with_callbacks = llm.bind(
    callbacks=[StdOutCallbackHandler()]
)

六、部署建议

1. 使用 FastAPI 封装：创建 RESTful API 接口
2. Docker 容器化：标准化部署环境
3. 异步处理：使用 Celery 处理长任务
4. 缓存策略：Redis 缓存热点查询结果
5. 监控告警：Prometheus + Grafana 监控系统

总结

LangChain 提供了完整的 LLM 应用开发生态，掌握其核心组件和最佳实践，可以帮助我们快速构建稳定可靠的企业级 AI 应用。

延伸阅读

• LangChain 官方文档
• LangChain 中文文档
• “Build a Large Language Model (From Scratch)”