首页 / 文章 / LangChain 1.x 实战：本地实现的链式结构、RAG、工具与智能体

LangChainRAGLLMOllamaPythonAgents

LangChain 1.x 实战：本地实现的链式结构、RAG、工具与智能体

无需API密钥，只需通过免费的本地Ollama环境即可使用LangChain 1.x学习构建链式结构、检索增强生成模型、相关工具以及代理型RAG系统。

3586 词

这是该实践系列的三部曲，前两部内容是介绍如何使用纯Python从零构建RAG和智能体。本篇采用的方法有所不同：您无需亲自组装每一个组件，而是会看到LangChain如何将同样的功能通过几行代码实现。由于您已经亲手构建过这些基础组件，因此能够更好地理解每种抽象层的具体作用，而不会将其视为神奇的魔法。这种理解非常重要——它决定了开发者是能够高效运用LangChain，还是不断与之斗争。

本教程中的所有操作都可以在本地免费运行，通过Ollama加载本地模型并使用本地嵌入技术。无需API密钥，也不存在速率限制的问题。

版本说明：本教程针对的是LangChain 1.x版本，已通过langchain==1.3.11和langchain-core==1.4.8进行验证。LangChain 1.0版本进行了重大重构——当前的智能体API以create_agent为核心，而诸如AgentExecutor和initialize_agent之类的旧组件则被移至独立的langchain-classic包中。网上许多教程仍在演示1.0版本之前的使用方式；此处展示的导入语句均为当前有效版本，且均已经过测试确认能够正确加载。

操作指南：打开名为 lc.py 的文件，按顺序运行每个代码块，并在遇到“轮到你了”类的练习时及时完成。每当看到“你已构建了此内容”这样的提示时，它指的是本系列前面教程中的手动实现方式。

步骤 0 — LangChain 到底是什么

LangChain 最恰当的理解是用于构建基于大语言模型应用程序的一系列标准化、可互换的组件——包括模型封装器、提示词模板、检索工具、向量存储、工具函数以及智能体等。这些组件都遵循统一的接口标准，因此你可以将它们相互连接，或用一个组件替换另一个（比如更换模型或向量存储），而无需重写应用程序的逻辑代码。

将所有内容串联起来的核心概念就是Runnable。每个组件都提供相同的.invoke()方法，且任意两个组件都可以通过|管道运算符连接起来。这种管道机制被称为LCEL，即LangChain表达式语言的缩写。一旦系统中的每个部分都遵循Runnable接口，那么整个RAG流程或智能体只需几行代码即可实现。

一个值得提前说明的权衡：LangChain能够减少重复代码，并提供大量现成的集成选项。但作为交换，它引入的抽象层可能会增加调试难度——有时你会更希望直接查看自己编写的原始循环代码。判断何时这种抽象层值得其带来的成本，才是真正的关键技能，我们将在第7步再次探讨这一权衡。

设置（免费的本地环境）

pip install langchain langchain-core langchain-ollama langchain-huggingface langchain-text-splitters sentence-transformers

你还需要安装Ollama（它是免费的且可在本地运行），之后应下载一个具备工具调用功能的模型：

ollama pull llama3.2     # ~2 GB; needs ~8 GB RAM. qwen2.5 also works well.

如果您不想使用 Ollama，仍然可以通过 langchain-huggingface 使用本地的 Hugging Face 模型来运行 chain 和 RAG 部分。不过，agent 部分依赖于可靠的工具调用功能，而那些计算能力有限的 CPU 模型往往难以满足这一需求。因此强烈建议在第 4 步到第 6 步中使用 Ollama。

第 1 步——核心操作：使用 | 管道构建链

在之前的 RAG 教程中，您是手动使用 f-string 构建提示词，将其传递给模型，然后通过 .strip() 方法处理输出结果。LangChain 则将相同的操作以管道表达式的形式实现。请将其添加到 lc.py 中：

from langchain_ollama import ChatOllama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOllama(model="llama3.2", temperature=0)

prompt = ChatPromptTemplate.from_template(
    "Explain {topic} in exactly one sentence."
)

# The chain: prompt -> model -> plain-string parser
chain = prompt | llm | StrOutputParser()

print(chain.invoke({"topic": "retrieval-augmented generation"}))

从左到右读取这个流程：prompt会将输入的字典转换为格式正确的消息，llm再将该消息转化为模型的响应，最后StrOutputParser()从响应对象中提取纯文本。

你其实已经实现过这个了。该流程在功能上与之前RAG演示中的f"Explain {topic}..."、generator(prompt)以及[0]["generated_text"].strip()完全相同——原本需要三步手动操作，现在则通过三个管道连接的Runnable实现。逻辑没有变化，只是接口得到了标准化。

现在轮到你了：每个Runnable都默认支持.batch()和.stream()方法。不妨试一试：

for piece in chain.stream({"topic": "vector embeddings"}):
    print(piece, end="", flush=True)   # tokens arrive as they're generated
print()
print(chain.batch([{"topic": "agents"}, {"topic": "chunking"}]))  # two at once

请注意，由于使用了标准的Runnable接口，流式处理和批量处理功能便免费随之提供了。这一“免费”优势实际上正是使用LangChain的全部意义所在。

第2步 —— LangChain方式的RAG

现在该用LangChain的构建模块来重新实现你之前手动编写的RAG流程了。每个组件都直接对应着你原先手写过的代码部分。

from langchain_huggingface import HuggingFaceEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Same Nimbus knowledge base from the RAG tutorial
DOCUMENTS = [
    "Nimbus is a fictional note-taking app. The free plan, Nimbus Lite, allows up to 50 notes and 1 GB of storage.",
    "Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed monthly. It includes 50 GB of storage and collaboration for up to 5 people.",
    "Nimbus stores notes encrypted at rest with AES-256. End-to-end encryption is Pro-only and must be enabled in Settings > Security.",
    "Nimbus offers a 30-day refund policy on all paid plans. Refunds reach the original payment method within 5 business days.",
    "Nimbus live chat support is staffed for Pro customers, Monday to Friday, 9am-6pm UTC. Free users get email support with a 48-hour response time.",
]

# 1. Split (↔ your chunk_text function)
splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=40)
chunks = splitter.create_documents(DOCUMENTS)

# 2. Embed locally (↔ your sentence-transformers model)
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")

# 3. Store + index (↔ your numpy array of vectors). No server needed.
vectorstore = InMemoryVectorStore.from_documents(chunks, embeddings)

# 4. Retriever (↔ your retrieve() with cosine top-k)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

for doc in retriever.invoke("How much does Pro cost?"):
    print("-", doc.page_content[:70], "...")

↔ 这全部都是你构建的。 RecursiveCharacterTextSplitter 扮演着文本分块器的角色，但它更加谨慎：它会依据段落和句子边界来分割文本，而不仅仅是统计单词数量。HuggingFaceEmbeddings 实际上就是对你之前使用的 all-MiniLM-L6-v2 模型的封装。InMemoryVectorStore 则对应着你用 numpy 创建的向量数组，其 .as_retriever() 方法会执行与你手动编写的相同的余弦相似度 top-k 搜索功能。这里仅四行代码就涵盖了你之前在步骤 2 到 4 中所实现的所有功能。

接下来，使用 LCEL 将检索功能与生成功能连接起来：

from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

rag_prompt = ChatPromptTemplate.from_template(
    "Answer using only the context. If it's not there, say you don't know.\n\n"
    "Context:\n{context}\n\nQuestion: {question}\nAnswer:"
)

def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()
)

print(rag_chain.invoke("How much does Nimbus Pro cost per month?"))

链式结构开头的字典会并行执行两个分支：question仅原样传递输入，而context则将相同输入传给检索器并格式化结果。两者输出随后都会进入提示语中。↔ 你构建的正是这个，本质上就是你之前的rag_answer()函数——检索、插入提示语、生成内容——全部浓缩在了一个表达式中。

轮到你了：试着调用 rag_chain.invoke("自由用户可以使用实时聊天吗？")，然后再调用一个无关的问题，比如 rag_chain.invoke("法国的首都是哪里？")。注意观察“我不知道”这样的回复——这与之前RAG教程中的基础检测相同，旨在强调同一个要点：检索的质量决定了答案的质量。之后，单独调用 retriever.invoke(...)，以便在回复出现异常时准确了解究竟检索到了什么内容。将检索与生成分开检查，这是你之前就常用的调试方法，而LangChain通过将这两个步骤作为独立的可执行任务来保留这一做法。

第3步 — 工具

在代理教程中，您将工具定义为 TOOLS 字典，并自行编写基于正则表达式的解析器，从模型的原始文本输出中提取工具名称及其输入参数。而 LangChain 通过原生工具调用功能完全省去了这种解析器的需求：您只需描述工具的功能，模型就会以结构化的调用形式响应，随后由 LangChain 负责路由处理。使用 @tool 装饰器定义工具的示例如下：

from langchain_core.tools import tool
import ast, operator, datetime

_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
        ast.Div: operator.truediv, ast.Pow: operator.pow, ast.USub: operator.neg}
def _ev(n):
    if isinstance(n, ast.Constant): return n.value
    if isinstance(n, ast.BinOp):   return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
    if isinstance(n, ast.UnaryOp): return _OPS[type(n.op)](_ev(n.operand))
    raise ValueError("unsupported")

@tool
def calculator(expression: str) -> str:
    """Evaluate a basic arithmetic expression like '8 * 12'."""
    return str(_ev(ast.parse(expression, mode="eval").body))

@tool
def get_today(_: str = "") -> str:
    """Return today's date in ISO format."""
    return datetime.date.today().isoformat()

以下是值得仔细查看的部分。您可以确切地了解 LangChain 是如何根据您的函数生成相应内容的：

print(calculator.name)         # 'calculator'
print(calculator.description)  # the docstring
print(calculator.args)         # {'expression': {'title': 'Expression', 'type': 'string'}}

最后那行是真实且经过验证的输出。LangChain检查了你的类型提示（expression: str）以及文档字符串，并据此构建了架构——模型正是通过这个架构来判断是否以及如何调用该工具。↔这是你构建的，之前你需要手动在SYSTEM_PROMPT中编写工具描述，再自行解析模型的输出。现在文档字符串本身就成为了描述，解析过程也会自动完成。这就解释了为什么文档字符串和类型提示至关重要——它们不仅仅是文档，还决定了模型如何理解和使用该工具。粗制滥造的文档字符串会导致模型错误地调用工具。

轮到你了：将计算器的文档字符串替换为毫无用处的内容，比如 """执行数学运算"""，然后再查看 .description 的内容。在第四步中，你将亲眼看到较弱的文档字符串如何导致模型做出更差的工具选择决策。你所编写的描述内容实际上起到了控制模型行为的作用。

第四步 —— 单次调用中的智能体

此前所做的努力在这里会显现成效。你自行构建的智能体需要循环结构、临时存储空间、解析器、错误处理机制、步骤限制，以及用于向模型传授 ReAct 格式的系统提示语。而在 LangChain 1.x 中，所有这些功能都整合到了一个函数调用中：create_agent。

from langchain.agents import create_agent

agent = create_agent(
    model=llm,                                   # your ChatOllama from Step 1
    tools=[calculator, get_today],               # the @tool functions from Step 3
    system_prompt="You are a helpful assistant. Use tools for math and dates.",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What is 8 times 12, and what is today's date?"}]}
)
print(result["messages"][-1].content)

这就是整个智能体的完整流程，从开始到结束。↔ 这一切都是你亲手构建的。 create_agent在内部会执行“推理-行动-观察”循环，将任务分配给相应的工具，把观察结果反馈给模型，检查停止条件并限制步骤数量——这些都是你在run_agent中手动实现的。其底层依赖于LangGraph，这也是该循环机制如此稳定的原因。

如果你想观察推理过程——也就是与设置verbose=True时获得的输出效果相同——可以实时查看中间步骤，而无需等待最终结果：

inputs = {"messages": [{"role": "user", "content": "How much is a year of Nimbus Pro?"}]}
for chunk in agent.stream(inputs, stream_mode="updates"):
    print(chunk)

在运行过程中，你会看到模型做出的每一个决策以及工具返回的每一个结果，它们会依次被打印出来。这与手动追踪时的思考/行动/观察模式完全相同，只不过是以结构化的更新对象形式呈现，而非需要你自己解析的原始文本。

轮到你了：试着提出一个需要模型将两个工具串联起来使用的题目——例如，假设试用今天开始，让模型计算30天退款期限的结束日期。观察它是否正确地依次调用get_today和calculator函数。之后，回到代理相关材料中的第7步——你在那里记录的所有故障模式（输出格式混乱、编造的工具名称、无限循环等）都可能在这里出现。该框架并不会提升弱模型的推理能力，只是隐藏了其背后的实现逻辑。理解这一区别正是为什么你比那些不先构建模型就直接使用框架的人能更快地调试这些代理。

第5步——整合功能：一个用于检索信息的代理（代理式RAG）

这就是前三节课内容交汇的地方。带上你的检索器，将其作为工具使用，然后将该工具交给智能体。从这时起，由智能体自行决定何时需要进行文档搜索——它可以自由地多次搜索，或根据需要将检索与计算相结合。

@tool
def search_nimbus_docs(query: str) -> str:
    """Search the Nimbus product documentation for facts about plans, pricing, refunds, security, and support."""
    docs = retriever.invoke(query)
    return "\n\n".join(d.page_content for d in docs)

smart_agent = create_agent(
    model=llm,
    tools=[search_nimbus_docs, calculator, get_today],
    system_prompt=(
        "You answer questions about the Nimbus app. "
        "Use search_nimbus_docs for any product facts, and calculator for arithmetic. "
        "Base answers only on retrieved facts."
    ),
)

q = "How much would Nimbus Pro cost a team of 4 for a full year?"
result = smart_agent.invoke({"messages": [{"role": "user", "content": q}]})
print(result["messages"][-1].content)

要正确回答这样的问题，智能体必须首先检索月度订阅价格，之后才进行8 * 12 * 4的计算。这就是将（第一课中的）检索功能以工具形式呈现（第三课），再由智能体驱动其执行——三个独立的概念整合成了一个系统。让智能体决定何时进行检索，比第二步中固定的rag_chain路径要灵活得多，而且这种模式在真实的生产系统中十分常见。

轮到您了：使用 smart_agent.stream(..., stream_mode="updates") 来实时展示该智能体的执行过程，并确认操作顺序——搜索应在计算之前进行。如果您的本地模型倾向于在脑海中直接求解算术题而非调用计算工具（这是小型模型的常见现象），请通过添加类似“您必须在每一步算术运算时都使用计算器”的要求来强化系统提示。这与智能体教程中使用的解决方法相同。

第6步——快速了解该工具包中的其他功能

至此，您已经搭建好了核心框架。还有几种LangChain的附加组件值得了解，它们每一项都对应着您之前手动构建过的功能：

  • 文档加载器（langchain-community）——可直接将PDF、网页、Notion页面等格式的资料导入到文本分割器已使用的相同Document对象中，从而替代手动将文本粘贴到列表中的方式，实现结构化的数据导入。
  • 生产级向量存储——可用Chroma或FAISS（通过from langchain_chroma import Chroma导入）替代内存存储，将嵌入向量持久化到磁盘上，从而突破内存限制实现更大规模的处理。由于这两种工具都提供了相同的.as_retriever()接口，因此链条中的后续组件无需做任何修改——这种一致性正是进行替换的核心目的。
  • 内存与消息历史记录——通过将对话链封装起来，使其能够保留之前轮次中的上下文，从而将单次交互的对话链转变为持续的对话。
  • 超越普通字符串的输出解析器——将模型的响应强制转换为 JSON 或经过验证的 Pydantic 对象，而非依赖原始文本恰好具有正确的格式。
  • LangGraph——当嵌入在 create_agent 中的循环逻辑不足以应对复杂场景时（如分支路径、需要人工审核的步骤、多个智能体协同工作等），就可以使用 LangGraph，这是 create_agent 所依赖的更低层级的图处理引擎。

    • 第 7 步——何时该使用 LangChain，何时不该用（坦诚分析）

    至此，你已经用两种方式构建了相同类型的系统——一次从零开始，一次借助框架——这让你有条件自行实现这一功能。这也是我们需要了解这两种版本的缘由。

    当你需要整合许多现有的组件——比如若干文档加载器、多个向量存储库以及多种模型提供方——并且不想为每个组件单独实现流处理、批量处理、重试逻辑和追踪功能时，LangChain就能发挥优势。此外，在你希望频繁更换组件且需要一个稳定的接口来背后处理这些切换，或是正在构建智能体而不愿自行维护推理循环时，LangChain也同样有用。

    当应用规模较小，学习 LangChain 的抽象概念所花费的时间会比直接编写那五十多行你已经熟悉的代码还要长时，手写代码往往是更好的选择。另外，在你需要完全了解正在执行的流程时——通过逐层调试框架来解决问题确实会带来很大麻烦，这种抱怨很有道理——或者当添加额外的间接层会掩盖本可以用纯 Python 更清晰表达的逻辑时，手写代码也是更优选择。在之前的教程中手动构建的 RAG 流水线和智能体完全适用于生产环境；使用框架并不会降低手写代码的价值。

    这里并没有唯一的正确答案。你首先学习手动版本，是为了在决定采用该框架时能够充分了解它将替代什么，从而做出明智的选择，而非因为其内部机制晦涩而不得不采用的默认方案。

    第8步 — 接下来该做什么

    • 如果本地环境的运行速度过慢，Groq和Google Gemini都提供免费托管的模型服务。切换操作非常简单——只需将ChatOllama替换为ChatGroq，或使用init_chat_model("gemini-...", model_provider="google_genai")即可，因为其他所有功能都基于相同的标准接口。无论选择哪种服务都需要API密钥，不过它们的免费套餐是完全免费的。
  • LangSmith 是 LangChain 的追踪与调试工具。当链或智能体行为异常时，它能让你查看每一步、每一个输入以及每一个输出。该工具提供免费版本，正好解决了“无法查看框架内部运作情况”这一痛点。
  • 当你需要带状态的工作流、多个协同工作的智能体，或是超出单一工具调用循环的人机交互检查点时，LangGraph 值得尝试。
  • 官方文档位于 docs.langchain.com。请务必确认你所阅读的内容对应的是 1.x 版本——1.0 之前的文档会提到 AgentExecutor、initialize_agent 或 LLMChain，这些名称如今都已更改或被废弃。

    • 值得牢记的思维模型

    LangChain本质上就是由你自行构建的组件，通过统一的接口Runnable进行标准化，并用|连接起来。一旦你亲自实现了这些组件，其中就没有任何真正新颖的概念了：

    • “链”其实就是你之前已经写好的从提示词到模型再到解析器的流程，只是将它们串联在了一起。
    • “检索器”就是你的嵌入向量与余弦相似度搜索逻辑，被封装在统一的接口中。
    • “工具”则是你编写的函数加上自动生成的架构，这样模型就可以直接调用它，而无需你再去解析其文本输出。
    • 通过create_agent创建的“智能体”，其实就是将你的整个推理-行动-观察循环整合在了一次调用中。

    当出现问题时，你可以用一贯的方法进行调试：找出出问题的组件。单独测试检索器，打印工具的.args信息，或查看代理的中间处理步骤。框架只是改变了你需要输入的代码量——并不会改变实际发生的情况，而你本来就已经明白正在发生什么。

    故障排除

    • 如果在调用create_agent或langchain_ollama时出现ImportError，很可能是使用了1.0版本之前的安装包或缺少某个依赖。请运行pip install -U langchain langchain-ollama，并确认langchain.__version__的值以1.开头。
  • 如果你正在阅读的教程中使用了 AgentExecutor 或 initialize_agent，那属于旧版 API。在 1.x 版本中，这些功能已被移至 langchain-classic；新代码应改用 create_agent。
  • Ollama 显示“连接被拒绝”的错误意味着服务器未运行——请使用 ollama serve 启动服务器或打开对应应用，并通过 ollama list 确认模型已显示。
  • 如果智能体忽略某些工具或试图自行进行算术运算，很可能是模型无法可靠地调用工具，这是小型模型的常见局限，也可能是工具的文档描述过于模糊。建议完善系统提示和工具的文档说明，或尝试使用更强大的模型如 qwen2.5。
  • HuggingFaceEmbeddings在首次使用时会显得较慢，因为它需要下载嵌入模型（约80 MB）并将其缓存到本地，之后进行检索的操作就会很快。
  • 通过Ollama首次调用模型时速度较慢，因为要将其加载到RAM中；后续的调用则很快。

    • 你现在已经亲手构建了RAG、智能体以及封装这两者的框架，先是手动实现，后来又使用了LangChain。你已了解了那些大多数人仅能从外部调用的层次结构。尽情去运用它吧。

    相关阅读

  • 将LLM作为法官式实时生产系统进行管理 — 了解Netflix的四阶段生命周期——真实数据、基于评分标准的训练、安全部署以及持续监控——如何在大规模应用中保持LLM法官系统的准确性。
    • 全部文章发送简报