LangChain 快速入门指南：从基础组件到服务部署

安装 LangChain

安装指定版本的 LangChain，建议安装最新稳定版。

pip install langchain langchain-openai

执行安装命令后，会自动安装相关组件，包括 langsmith、langchain-core、langchain-text-splitters 等。

更新 LangChain：

pip install --upgrade langchain

克隆 LangChain 存储库从源代码安装：

pip install -e .

安装 LangChain 时包括常用的开源 LLM（大语言模型）库：

pip install langchain[llms]

配置环境变量

安装第三方集成库，以使用 OpenAI。需要设置环境变量以认证请求。

import os

# 设置 OpenAI 基础 URL
os.environ["OPENAI_BASE_URL"] = "https://api.openai.com/v1"
# 设置 OpenAI API Key，请替换为您的真实密钥
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)

虽然模型依然不知道具体答案，但对于技术作者来说，它使用了更恰当的语气给予回应。

使用输出解析器

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

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
response = chain.invoke({"input": "LangSmith 如何帮助进行测试？"})
print(response)

向量存储

加载要索引的数据，需要安装 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!")]
result = retriever_chain.invoke({
    "chat_history": chat_history,
    "input": "告诉我怎么做"
})
print(result)

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

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!")]
result = retrieval_chain.invoke({
    "chat_history": chat_history,
    "input": "Tell me how"
})
print(result["answer"])

代理的使用

构建代理时要做的第一件事是确定它应该有权访问哪些工具。这里授予代理访问两个工具的权限：使用创建的检索器，以便代理能够回答有关 LangSmith 的问题；以及一个搜索工具，以使代理能够回答需要最新信息的问题。

检索器工具

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
result = agent_executor.invoke({"input": "Langsmith 如何帮助进行测试？"})
print(result["output"])

询问天气情况：

# 执行 Agent
result = agent_executor.invoke({"input": "成都今天天气情况？"})
print(result["output"])

进行对话：

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

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

LangServe 提供服务

概述

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

安装 langserve：

pip install "langserve[all]"

创建服务

创建一个 serve.py 文件。包含为应用程序提供服务的逻辑。由三部分组成：构建的链的定义、FastAPI 应用程序、为链提供服务的路由的定义，由 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. 获取检索器
# 创建一个 WebBaseLoader 对象，加载给定 URL 的网页内容
loader = WebBaseLoader("https://docs.smith.langchain.com/user_guide")
# 载入网页内容
docs = loader.load()

# 初始化 RecursiveCharacterTextSplitter 对象用于文本拆分
text_splitter = RecursiveCharacterTextSplitter()
# 使用文本拆分器将文档分成段落
documents = text_splitter.split_documents(docs)

# 初始化 OpenAIEmbeddings 对象，用于获取文本嵌入
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. 创建代理
# 从指定的 Hub 拉取提示模板
prompt = hub.pull("hwchase17/openai-functions-agent")

# 初始化 ChatOpenAI 对象，选择模型为"gpt-3.5-turbo"，设置温度为 0
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 使用提供的模型、工具和提示创建 OpenAI 函数代理器
agent = create_openai_functions_agent(llm, tools, prompt)

# 初始化 AgentExecutor，传入代理器、工具对象和 verbose 标记为 True
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):
    # 定义输入 BaseModel 包含字段 input 和 chat_history
    input: str
    chat_history: List[BaseMessage] = Field(
        ...,
        extra={"widget": {"type": "chat", "input": "location"}}
    )

class Output(BaseModel):
    # 定义输出 BaseModel 包含字段 output
    output: str

# 将该配置的 agent_executor 添加到应用程序 app 的路由中，路径为 "/agent"
add_routes(
    app,
    # agent_executor 配置为使用特定的输入和输出类型
    agent_executor.with_types(input_type=Input, output_type=Output),
    path="/agent",
)

if __name__ == "__main__":
    # 导入 uvicorn 模块
    # uvicorn 是用于 ASGI 应用程序的轻量级 Web 服务器
    import uvicorn

    # 运行主应用程序 app，指定主机为 localhost，端口为 8000
    uvicorn.run(app, host="localhost", port=8000)

启动服务

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

python serve.py

启动成功后，控制台会显示服务运行信息，并提示 Playground 地址。

服务交互

每个 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)

最佳实践与注意事项

1. 安全性：在生产环境中，切勿将 API Key 硬编码在代码中。建议使用环境变量或密钥管理服务。
2. 性能优化：向量存储的索引大小会影响检索速度，定期清理过期文档。
3. 错误处理：在 Agent 执行过程中，应捕获可能的异常，防止服务崩溃。
4. 版本管理：LangChain 迭代迅速，建议关注官方发布日志，及时更新依赖以避免兼容性问题。