LangChain学习笔记：从聊天模型到AI Agent的进阶之路

本文将分享我近几周来学习 LangChain 的心得体会，从基础的 聊天模型 构建，到复杂的 检索增强生成 (RAG) 系统 和 AI Agent 的实现，希望能为正在学习 LangChain 的朋友提供一些参考。LangChain 作为一个强大的框架，极大地简化了基于 大模型 的应用开发流程，让我们能够更高效地构建各种智能应用。

1. 从零开始：掌握 LangChain 聊天模型

我首先从 LangChain 的 聊天模型 入手。基础的 聊天模型 设置非常简单，只需要几行代码就能实现与 大模型 的交互。

from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()  # 加载 .env 文件中的环境变量

llm = ChatOpenAI(model="gpt-4o") #指定模型

result = llm.invoke("法国的首都是什么？") # 提问

print(result.content) # 输出回答

关键学习： 务必使用环境变量来管理 API 密钥。load_dotenv() 函数可以自动从 .env 文件中加载 OpenAI API 密钥，确保敏感信息的安全。这不仅是最佳实践，也能避免硬编码密钥带来的安全风险。例如，在团队协作开发时，将 API 密钥直接写入代码会导致泄露风险，而使用 .env 文件可以方便地为每位开发者配置不同的密钥。

除了简单的问答，LangChain 还提供了不同的消息类型，用于构建更结构化的对话。例如，SystemMessage 可以用来设定 AI Agent 的角色和行为准则。

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_openai import ChatOpenAI

messages = [
    SystemMessage(content="你是一个乐于助人的助手，可以回答问题并帮助创建社交媒体内容。"),
    HumanMessage(content="给出一个在 LinkedIn 上创建吸引人帖子的简短技巧")
]

llm = ChatOpenAI(model="gpt-4o")
result = llm.invoke(messages)
print(result.content)

关键学习： 使用 SystemMessage 设置上下文对于确保 AI Agent 行为的一致性至关重要。它相当于为你的 AI 助手设定了 “个性” 或 “角色” 指令。如果想要让 AI 助手更专业，可以指定它为“资深营销专家”；如果想要更活泼，可以设置为“幽默风趣的段子手”。

更进一步，可以构建一个具有记忆功能的对话式聊天机器人。

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")
chat_history = []
system_message = SystemMessage(content="你是一个乐于助人的助手。")
chat_history.append(system_message)

while True:
    user_input = input("You: ")
    if user_input.lower() == "exit":
        break
    chat_history.append(HumanMessage(content=user_input))
    result = model.invoke(chat_history)
    response = result.content
    chat_history.append(AIMessage(content=response))
    print(f"AI: {response}")

关键学习： 维护聊天记录对于上下文感知的对话至关重要。每次交互都建立在之前的消息之上，从而创建自然的对话流程。想象一下，如果每次对话 AI 都要重新了解你的需求，那体验将会非常糟糕。维护聊天记录就像是给 AI 配备了 “短期记忆”，让它能更好地理解你的意图。

2. 灵活应变：掌握 Prompt 模板的艺术

仅仅是简单的对话是不够的，我们需要让 大模型 能够根据不同的输入生成不同的内容。Prompt 模板 正是为了解决这个问题而生的。它允许你创建可重用的、参数化的提示词。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

messages = [
    ("system", "你是一个乐于助人的助手，可以帮你讲关于{topic}的笑话。"),
    ("user", "给我讲{number}个关于{topic}的笑话")
]

prompt_template = ChatPromptTemplate.from_messages(messages)
prompt = prompt_template.invoke({
    "number": 2,
    "topic": "人工智能"
})

llm = ChatOpenAI(model="gpt-4o")
result = llm.invoke(prompt)
print(result.content)

关键学习： 模板使你的提示词可重用且易于维护。与其硬编码提示词，不如创建灵活的模板，使其能够适应不同的输入。例如，你可以创建一个通用的“产品描述生成器”模板，只需要传入产品名称、特点和目标受众，就能自动生成高质量的产品描述。根据 Databox 的一项调查，使用模板可以节省 50% 以上的内容创作时间。

3. 融会贯通：构建强大的 LangChain Chains

Chains 是 LangChain 连接不同组件的方式。一个简单的 Chain 如下所示：

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

model = ChatOpenAI(model="gpt-4o")

prompt_template = ChatPromptTemplate.from_messages([
    ("system", "你是一位事实专家，了解关于{topic}的事实"),
    ("user", "给我{number}个关于{topic}的事实")
])

# 管道操作符 (|) 将组件连接在一起
chain = prompt_template | model | StrOutputParser()

result = chain.invoke({
    "topic": "人工智能",
    "number": 5
})

print(result)

关键学习： 管道操作符 | 是 LangChain 连接组件的方式。数据从左到右流经每个组件。这使得构建复杂的流程变得非常简单，就像搭积木一样。例如，可以将一个 Chain 设计为“从用户输入提取关键词 -> 调用搜索引擎 -> 从搜索结果中提取关键信息 -> 生成摘要”。

更高级的 Chain 可以包含自定义逻辑：

from langchain_core.runnables import RunnableLambda, RunnableSequence

format_prompt = RunnableLambda(lambda x: prompt_template.format_prompt(**x))
invoke_model = RunnableLambda(lambda x: model.invoke(x.to_messages()))
parse_output = RunnableLambda(lambda x: x.content)

chain = RunnableSequence(
    first=format_prompt,
    middle=[invoke_model],
    last=parse_output
)

关键学习： RunnableLambda 允许你将自定义逻辑注入到 Chains 中，为复杂的数据转换提供灵活性。这使得你可以根据具体需求对数据进行预处理、后处理或者进行复杂的业务逻辑计算。

Chains 还可以执行复杂的多步骤操作，比如生成内容并将其翻译成另一种语言：

translation_template = ChatPromptTemplate.from_messages([
    ("system", "你是一位乐于助人的助手，可以将文本翻译成{language}"),
    ("human", "将以下文本翻译成{language}：{text}")
])

prepare_for_translation = RunnableLambda(
    lambda output: {"text": output, "language": "French"}
)

chain = (prompt_template | model | StrOutputParser() |
         prepare_for_translation | translation_template |
         model | StrOutputParser())

关键学习： Chains 可以执行复杂的多步骤操作。每个步骤的输出都成为下一步骤的输入。这就像一个流水线，每个环节都负责不同的任务，最终完成整个流程。

对于可以同时运行的操作，可以使用 RunnableParallel 来提高性能：

from langchain_core.runnables import RunnableParallel

def analyze_plot(plot):
    # 自定义分析逻辑
    return plot_template.format_prompt(plot=plot)

def analyze_characters(characters):
    # 自定义分析逻辑
    return characters_template.format_prompt(characters=characters)

plot_branch_chain = RunnableLambda(analyze_plot) | model | StrOutputParser()
characters_branch_chain = RunnableLambda(analyze_characters) | model | StrOutputParser()

chain = (
    summary_template | model | StrOutputParser()
    | RunnableParallel(branches={
        "plot": plot_branch_chain,
        "characters": characters_branch_chain
    })
    | RunnableLambda(lambda x: combine_results(
        x["branches"]["plot"],
        x["branches"]["characters"]
    ))
)

关键学习： RunnableParallel 启用并发处理，显著提高了独立操作的性能。例如，在分析一篇长篇文章时，可以将“情感分析”、“关键词提取”和“实体识别”等任务并行执行，从而大大缩短处理时间。根据 Intel 的一项研究，并行处理可以将某些任务的性能提升 2-4 倍。

RunnableBranch 允许在 Chains 中实现条件逻辑：

from langchain_core.runnables import RunnableBranch

classification_chain = classification_template | model | StrOutputParser()

branches = RunnableBranch(
    (lambda x: "positive" in x, positive_feedback_template | model | StrOutputParser()),
    (lambda x: "negative" in x, negative_feedback_template | model | StrOutputParser()),
    (lambda x: "neutral" in x, neutral_feedback_template | model | StrOutputParser()),
    escalation_template | model | StrOutputParser()  # 默认情况
)

chain = classification_chain | branches

关键学习： RunnableBranch 启用 Chains 中的条件逻辑，允许根据输入特征采用不同的处理路径。这使得我们可以根据用户的反馈（例如，正面、负面或中性）采取不同的行动，从而构建更智能、更灵活的系统。

4. 知识引擎：构建强大的 RAG 系统

RAG (Retrieval-Augmented Generation) 是我学到的最令人兴奋的概念之一。它结合了检索和生成的能力，使得 大模型 能够利用外部知识来生成更准确、更丰富的答案。

首先，需要设置一个向量数据库：

import os
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import CharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

# 加载并分割文档
file_path = os.path.join("documents", "AAW.txt")  # 《爱丽丝梦游仙境》
loader = TextLoader(file_path, encoding='utf-8')
documents = loader.load()
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
docs = text_splitter.split_documents(documents)

# 创建嵌入和向量存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
db = Chroma.from_documents(docs, embeddings, persist_directory="db/chroma_db")

关键学习： 文档分块对于 RAG 系统 至关重要。chunk_size 和 chunk_overlap 参数需要根据你的内容类型进行调整。如果块太大，可能会包含太多无关信息；如果块太小，可能会丢失上下文信息。通常需要进行实验，找到最佳的参数组合。

接下来，可以查询向量数据库：

# 设置具有相似性搜索的检索器
retriever = db.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "k": 3,  # 返回前 3 个最相似的块
        "score_threshold": 0.5  # 最小相似度分数
    })

query = "《爱丽丝梦游仙境》是关于什么的？"
relevant_docs = retriever.invoke(query)

for i, doc in enumerate(relevant_docs, 1):
    print(f"Document {i}:\n{doc.page_content}\n")
    print(f"Source: {doc.metadata.get('source', 'Unknown')}\n")

关键学习： 相似度阈值和结果数量 (k) 是影响检索质量的重要参数。太低的阈值可能会包含不相关的内容；太高的阈值可能会错过相关信息。合理的设置能保证检索结果的准确性和覆盖率。

最后，可以将检索与生成结合起来，构建完整的 RAG 实现：

from langchain.schema import HumanMessage, SystemMessage

query = "《爱丽丝梦游仙境》中的主要角色是什么？"
relevant_docs = retriever.invoke(query)

# 将检索到的文档与查询结合起来
combined_input = (
    f"这里有一些可能有助于回答问题的文档：{query}\n\n"
    "相关文档：\n" +
    "\n\n".join([doc.page_content for doc in relevant_docs]) +
    "\n\n请仅根据提供的文档提供答案。"
    "如果在文档中找不到答案，请回答“我不确定”。"
)

model = ChatOpenAI(model="gpt-4o")
messages = [
    SystemMessage(content="你是一位乐于助人的助手，可以根据提供的文档回答问题。"),
    HumanMessage(content=combined_input)
]

result = model.invoke(messages)
print(result.content)

关键学习： RAG 系统 擅长提供基于事实的、基于来源的答案。关键在于明确指示模型仅依赖提供的文档，并在信息不可用时承认不确定性。这可以有效避免 大模型 产生幻觉，提高答案的可靠性。

添加元数据可以更好地组织文档：

# 加载具有元数据的多个文档
documents = []
for book_file in book_files:
    file_path = os.path.join(books_dir, book_file)
    loader = TextLoader(file_path, encoding="utf-8")
    book_docs = loader.load()

    for doc in book_docs:
        doc.metadata = {"source": book_file}
        documents.append(doc)

关键学习： 元数据对于跟踪文档来源和启用过滤搜索至关重要。它可以帮助用户了解信息的来源。例如，可以添加“作者”、“发布日期”、“主题”等元数据，方便用户根据不同的维度进行检索。

5. 智能助手：创建 AI Agent

AI Agent 可以使用工具来执行特定任务。下面是一个用于获取时间的自定义工具：

from langchain.agents import tool
import datetime
import pytz

@tool
def get_current_time(timezone_name: str) -> str:
    """获取指定时区的当前时间"""
    try:
        timezone = pytz.timezone(timezone_name)
        current_time = datetime.datetime.now(timezone)
        return current_time.strftime('%Y-%m-%d %H:%M:%S %Z')
    except:
        return "无效时区"

关键学习： @tool 装饰器自动从任何函数创建 LangChain 兼容的工具。文档字符串成为工具描述，帮助 AI Agent 了解何时使用它。清晰的工具描述至关重要，可以帮助 AI Agent 做出更明智的决策。

创建一个 ReAct Agent:

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

model = ChatOpenAI(model="gpt-4o")
prompt_template = hub.pull("hwchase17/react")  # 预构建的 ReAct 提示词

tools = [get_current_time]
agent = create_react_agent(model, tools, prompt_template)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({
    "input": "斯里兰卡和伦敦的当前时间是多少？"
})

关键学习： ReAct (Reasoning + Acting) Agent 遵循一种思考-行动-观察模式。他们思考要做什么，使用工具采取行动，观察结果，然后继续直到他们找到答案。这种模式使得 AI Agent 能够处理复杂的任务，并从错误中学习。

6. 总结与最佳实践

通过这段时间的学习，我对 LangChain 有了更深入的理解。以下是一些关键的学习和最佳实践：

技术层面：

环境管理： 始终使用环境变量来管理 API 密钥和敏感配置。
错误处理： 实施适当的错误处理，特别是对于外部 API 调用和文件操作。
性能： 当操作独立时，使用并行处理 (RunnableParallel)。
内存管理： 在构建对话历史记录时，注意令牌限制。

设计模式：

模块化架构： 将复杂的 Chains 分解为更小、可重用的组件。
模板可重用性： 创建可在不同上下文中重用的灵活提示模板。
关注点分离： 保持数据加载、处理和生成逻辑分离。

RAG 最佳实践：

分块大小优化： 根据你的内容类型试验不同的分块大小。
相似度阈值： 调整相似度阈值以平衡相关性和覆盖率。
元数据使用： 包括丰富的元数据，以便更好地跟踪和过滤文档。

Agent 设计：

工具描述： 编写清晰、详细的工具描述，以帮助 Agent 做出更好的决策。
错误恢复： 设计工具以优雅地处理错误并提供有意义的反馈。
范围限制： 为 Agent 可以做什么和不能做什么定义明确的边界。

总而言之，LangChain 的模块化设计和强大的抽象使得构建复杂的系统成为可能，同时保持了干净、可读的代码。无论你是在构建聊天机器人、知识管理系统还是 AI Agent，LangChain 都提供了你需要的基础。希望本文能帮助你更好地理解和使用 LangChain，构建出更强大的 大模型 应用。

LangChain学习笔记：从聊天模型到AI Agent的进阶之路