LangChain 快速上手指南：核心组件与实战应用

此外，安装 LangChain 时可包含常用的开源 LLM 库：

pip install langchain[llms]

配置环境变量

安装第三方集成库以使用 OpenAI 服务：

pip install langchain langchain_openai

设置 OpenAI 环境变量。注意：请勿将真实 API Key 提交至代码仓库。

import os

os.environ["OPENAI_BASE_URL"] = "https://api.openai.com/v1"
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

基本使用示例

初始化模型并尝试调用：

from langchain_openai import ChatOpenAI

# 初始化模型
llm = ChatOpenAI()

# 调用模型
response = llm.invoke("LangSmith 是什么？")
print(response.content)

虽然模型可能无法准确回答特定专有名词，但展示了基本的交互流程。

使用提示模板

使用提示模板来指导模型响应。提示模板将原始用户输入转换为更适合 LLM 理解的格式。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 初始化模型
llm = ChatOpenAI()

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "您是世界级的技术文档编写者。"),
    ("user", "{input}")
])

# 组合成一个简单的 LLM 链
chain = prompt | llm 

# 使用 LLM 链
response = chain.invoke({"input": "LangSmith 如何帮助进行测试？"})
print(response.content)

通过添加 System Prompt，模型会采用更专业的语气进行回应。

使用输出解析器

添加一个简单的输出解析器，将聊天消息转换为字符串，便于后续处理。

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

# 初始化模型
llm = ChatOpenAI()

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "您是世界级的技术文档编写者。"),
    ("user", "{input}")
])

# 使用输出解析器
output_parser = StrOutputParser()

# 将其添加到上一个链中
chain = prompt | llm | output_parser

# 调用它并提出同样的问题。答案是一个字符串，而不是 ChatMessage
result = chain.invoke({"input": "LangSmith 如何帮助进行测试？"})
print(result)

向量存储与 RAG

加载要索引的数据，需要安装 BeautifulSoup 用于网页解析：

pip install beautifulsoup4

将其索引到向量存储中。这需要嵌入模型和向量存储组件。

使用一个简单的本地向量存储 FAISS，首先需要安装它。

pip install faiss-cpu

代码实现：

# 导入和使用 WebBaseLoader
from langchain_community.document_loaders import WebBaseLoader

loader = WebBaseLoader("https://docs.smith.langchain.com/user_guide")
docs = loader.load()

# 对于嵌入模型，这里通过 API 调用
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()

# 使用此嵌入模型将文档摄取到矢量存储中
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 使用分割器分割文档
text_splitter = RecursiveCharacterTextSplitter()
documents = text_splitter.split_documents(docs)

# 向量存储
vector = FAISS.from_documents(documents, embeddings)

检索链

已在向量存储中索引了这些数据，接下来创建一个检索链。该链将接收一个传入的问题，查找相关文档，然后将这些文档与原始问题一起传递给 LLM，要求它回答原始问题。

from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("""仅根据提供的上下文回答以下问题:

<context>
{context}
</context>

Question: {input}""")

# 创建链，该链获取文档列表并将它们全部格式化为提示，然后将该提示传递给 LLM。
# 它传递所有文档，因此应该确保它适合正在使用的 LLM 上下文窗口
document_chain = create_stuff_documents_chain(llm, prompt)

# 可以直接通过传入文档来运行它
from langchain_core.documents import Document

text = "langsmith can let you visualize test results"
document_chain.invoke({
    "input": "LangSmith 如何帮助进行测试？",
    "context": [Document(page_content=text)]
})

还可以让文档首先来自刚刚设置的检索器。这样，可以使用检索器动态选择最相关的文档，并将其传递给给定的问题。

from langchain.chains import create_retrieval_chain

# 创建向量存储检索器
retriever = vector.as_retriever()
# 创建链，该链接收用户查询，然后将其传递给检索器以获取相关文档。
# 然后将这些文档（和原始输入）传递到 LLM 以生成响应
retrieval_chain = create_retrieval_chain(retriever, document_chain)
# 执行检索 这将返回一个字典
response = retrieval_chain.invoke({"input": "how can langsmith help with testing?"})
print(response["answer"])

对话检索链

上面创建的链只能回答单个问题。现在创建一个新链。该链将接收最新的输入和对话历史记录，并使用 LLM 生成搜索查询。

from langchain.chains import create_history_aware_retriever
from langchain_core.prompts import MessagesPlaceholder

# First we need a prompt that we can pass into an LLM to generate this search query
prompt = ChatPromptTemplate.from_messages([
    MessagesPlaceholder(variable_name="chat_history"),
    ("user", "{input}"),
    ("user", "鉴于上述对话，生成一个搜索查询以查找以获取与对话相关的信息")
])
retriever_chain = create_history_aware_retriever(llm, retriever, prompt)

# 通过传入用户提出后续问题来测试
from langchain_core.messages import HumanMessage, AIMessage

chat_history = [HumanMessage(content="LangSmith 可以帮助测试我的 LLM 应用程序吗？"), AIMessage(content="Yes!")]
retriever_chain.invoke({
    "chat_history": chat_history,
    "input": "告诉我怎么做"
})

还可以创建一个新的链来继续对话，并牢记这些检索到的文档。

prompt = ChatPromptTemplate.from_messages([
    ("system", "根据以下上下文回答用户的问题:\n\n{context}"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("user", "{input}"),
])

document_chain = create_stuff_documents_chain(llm, prompt)
retrieval_chain = create_retrieval_chain(retriever_chain, document_chain)

# 测试
chat_history = [HumanMessage(content="LangSmith 可以帮助测试我的 LLM 应用程序吗？"), AIMessage(content="Yes!")]
retrieval_chain.invoke({
    "chat_history": chat_history,
    "input": "Tell me how"
})

代理的使用

构建代理时要做的第一件事是确定它应该有权访问哪些工具。这里授予代理访问两个工具的权限：

1. 使用创建的检索器，以便代理能够回答有关 LangSmith 的问题。
2. 一个搜索工具，以使代理能够回答需要最新信息的问题。

检索器工具

from langchain.tools.retriever import create_retriever_tool

retriever = vector.as_retriever()
retriever_tool = create_retriever_tool(
    retriever,
    "langsmith_search",
    "搜索有关 LangSmith 的信息。对于有关 LangSmith 的任何问题，您必须使用此工具！",
)

创建搜索工具

访问 Tavily 注册账号登录并创建 API 秘钥，然后配置环境变量。

import os
os.environ["TAVILY_API_KEY"] = "YOUR_TAVILY_API_KEY"

安装 tavily-python 库：

pip install -U langchain-community tavily-python

创建工具：

from langchain_community.tools.tavily_search import TavilySearchResults
search = TavilySearchResults()

创建代理

创建使用工具的列表：

tools = [retriever_tool, search]

创建一个代理来使用工具：

from langchain_openai import ChatOpenAI
from langchain import hub
from langchain.agents import create_openai_functions_agent
from langchain.agents import AgentExecutor

# 获取使用提示 可以修改它
prompt = hub.pull("hwchase17/openai-functions-agent")
# 初始化大模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 创建一个 openai_functions_agent 代理
agent = create_openai_functions_agent(llm, tools, prompt)
# 创建代理执行器
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

使用 Agent 代理

# 执行 Agent
agent_executor.invoke({"input": "Langsmith 如何帮助进行测试？"})

询问天气情况：

# 执行 Agent
agent_executor.invoke({"input": "成都今天天气情况？"})

进行对话：

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

chat_history = [HumanMessage(content="LangSmith 可以帮助测试我的 LLM 应用程序吗？"), AIMessage(content="Yes!")]
agent_executor.invoke({
    "chat_history": chat_history,
    "input": "告诉我怎么做"
})

LangServe 提供服务

概述

LangServe 可以帮助开发人员将 LangChain 应用程序部署为 REST API。使用 LangChain 时不是必定使用 LangServe，但在生产环境中推荐使用。

安装 langserve：

pip install "langserve[all]"

创建服务

创建一个 serve.py 文件。包含为应用程序提供服务的逻辑。由三部分组成：

1. 构建的链的定义
2. FastAPI 应用程序
3. 为链提供服务的路由的定义，由 langserve.add_routes 命令完成

import os
from typing import List

from fastapi import FastAPI
from langchain import hub
from langchain.agents import AgentExecutor
from langchain.agents import create_openai_functions_agent
from langchain.pydantic_v1 import BaseModel, Field
from langchain.tools.retriever import create_retriever_tool
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_community.vectorstores import FAISS
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langserve import add_routes

# 配置环境变量
os.environ["TAVILY_API_KEY"] = "YOUR_TAVILY_API_KEY"
os.environ["OPENAI_BASE_URL"] = "https://api.openai.com/v1"
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 1. 获取检索器
loader = WebBaseLoader("https://docs.smith.langchain.com/user_guide")
docs = loader.load()

text_splitter = RecursiveCharacterTextSplitter()
documents = text_splitter.split_documents(docs)

embeddings = OpenAIEmbeddings()
vector = FAISS.from_documents(documents, embeddings)
retriever = vector.as_retriever()

# 2. 创建工具
retriever_tool = create_retriever_tool(
    retriever,
    "langsmith_search",
    "Search for information about LangSmith. For any questions about LangSmith, you must use this tool!",
)

search = TavilySearchResults()
tools = [retriever_tool, search]

# 3. 创建代理
prompt = hub.pull("hwchase17/openai-functions-agent")
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 4. 应用定义
app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="A simple API server using LangChain's Runnable interfaces",
)

# 5. 添加路由
class Input(BaseModel):
    input: str
    chat_history: List[BaseMessage] = Field(
        ...,
        extra={"widget": {"type": "chat", "input": "location"}}
    )

class Output(BaseModel):
    output: str

add_routes(
    app,
    agent_executor.with_types(input_type=Input, output_type=Output),
    path="/agent",
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

启动服务

执行这个文件启动服务，并在 localhost:8000 上提供服务：

python serve.py

启动成功后，可以看到服务运行在 http://localhost:8000。

服务交互

每个 LangServe 服务都带有一个简单的内置 UI，用于配置和调用具有流输出和中间步骤可见性的应用程序。 访问：http://localhost:8000/agent/playground/

设置一个客户端，以便以编程方式与我们的服务进行交互：

from langserve import RemoteRunnable

remote_chain = RemoteRunnable("http://localhost:8000/agent/")
res = remote_chain.invoke({
    "input": "成都今天天气情况怎样？",
    "chat_history": []
})
print(res)

最佳实践与总结

在实际开发中，为了确保 LangChain 应用的稳定性和性能，建议遵循以下最佳实践：

1. 错误处理：在生产环境中，务必对 LLM 调用的超时、网络异常以及 API 限流进行捕获和处理，避免服务中断。
2. 缓存机制：对于重复的查询或相同的上下文，可以考虑引入 Redis 等缓存层，减少不必要的 Token 消耗和延迟。
3. 隐私安全：切勿将真实的 API Key 硬编码在代码中，应使用环境变量或密钥管理服务（如 AWS Secrets Manager）。
4. 上下文长度管理：在使用 RAG 时，注意控制检索到的文档总长度，避免超出 LLM 的最大上下文窗口限制。
5. 监控与日志：集成 LangSmith 或其他监控工具，记录每次请求的输入、输出及耗时，便于排查问题和优化体验。

通过本文的介绍，您已经掌握了 LangChain 的核心组件、基础用法、RAG 实现、Agent 构建以及服务部署的全流程。LangChain 作为一个强大的生态框架，极大地降低了构建 AI 应用的门槛。随着技术的演进，建议持续关注官方文档和社区动态，探索更多高级功能，如多模态支持、复杂工作流编排等，以满足日益增长的业务需求。