LangChain项目代码拆解：Chain、Agent、Memory与Retrieval四大核心组件实战指南

简介：这是一份LangChain入门与实践的项目代码包，面向正在学习大模型应用开发的Python开发者，旨在帮助读者快速理解LangChain的模块化设计与核心用法。包内共3个文件，包含一个HTML格式的入门讲解页、一个项目运行配置文件以及一个版本管理忽略文件，压缩后仅5KB，体量非常轻巧。LangChain作为连接LLM与外部数据、工具的开源框架，其核心模块涉及提示词模板、链式操作、记忆功能和工具调用等，读者可借助这份代码包中的HTML页面快速建立对框架结构的整体认知，并根据配置信息了解项目的基础组织方式，进而结合官方文档动手实践。该代码包对应的实践内容可迁移至智能问答、对话机器人等常见落地场景，帮助初学者在极低时间成本下感受LangChain的典型用法。目前已有107人学习，文件虽少但聚焦核心，适合用极低成本开启LangChain的代码初体验。

1. 拿到《LangChain入门与实践》项目代码，别直接跑：先看清四条主线再动手

拿到一份《LangChain入门与实践》项目代码，最容易被看出外行的动作是翻完 README 就直接跑 demo。LangChain 版本迭代速度很快，代码里很多写法在你本地环境里可能已经换新，跑通只是第一关，真正值钱的是看懂 Chain、Agent、Memory、Retrieval 这四条主线在代码里是怎么被编排的。这篇文章要解决的不只是怎么运行，而是沿着项目代码的结构，把每条主线对应的实现、参数和常见翻车点讲清楚。适合刚写完一轮 Prompt 想开始做应用的开发者，也适合准备在团队里搭 LLM 工具链、需要拆别人代码找思路的同学。

2. 拆开项目代码外壳：先认准 Chain、Agent、Memory、Retrieval 四根柱子

LangChain 的入门项目代码通常会按功能拆出几个脚本，看起来文件很多，追一条请求进去就能发现，所有逻辑都围绕四类组件：Chain 负责编排一次或多次调用，Agent 负责在循环里自主决策，Memory 负责跨轮记忆，Retrieval 负责把外部知识带进 Prompt。先认准它们，读代码就不会迷路。

2.1 最小 Chain：一条 Prompt 通过管道符变成一次模型调用

Chain 是理解 LangChain 的起点。最精简的一条链是 Prompt 加上模型，再加上输出解析器。项目代码里的 chain_demo.py 往往长这样。

# chain_demo.py from langchain_core.language_models import BaseChatModel from langchain_core.messages import AIMessage from langchain_core.outputs import ChatGeneration, ChatResult from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate class FakeChatModel(BaseChatModel): """离线模拟模型：不连外部服务，先把链路跑通。""" def _generate(self, messages, stop=None, run_manager=None, **kwargs): last = messages[-1].content return ChatResult(generations=[ ChatGeneration(message=AIMessage( content=f"这是离线模拟回答，最后一段消息是：{last[:20]}" )) ]) @property def _llm_type(self): return "fake-chat-model" prompt = ChatPromptTemplate.from_template("请用一句话解释：{topic}") fake_llm = FakeChatModel() chain = prompt | fake_llm | StrOutputParser() print(chain.invoke({"topic": "LangChain"}))

这段代码里的管道符|是 LangChain Expression Language 的串联语法，也可以叫 LCEL。执行到chain.invoke时，LangChain 先把{"topic": "LangChain"}填进 Prompt 模板，再把格式化后的消息发给模型，拿到ChatResult后由StrOutputParser转成纯字符串返回。FakeChatModel继承的是BaseChatModel，它强制实现_generate方法，这是所有 ChatModel 的统一入口，真实项目里你只需要把FakeChatModel替换成你接入的模型服务商提供的驱动类，链的其余代码不用改。

BaseChatModel抽象的价值在于把“换模型”的成本降到最低。你更换供应商时，只改驱动类实例化那一行，后面的 Prompt、解析器、调用方式全部保持一致。入门项目代码里用这种抽象，目的就是让你先忽略供应商差异，专注理解结构。

如果想让这条链支持流式输出，可以调用chain.stream({"topic": "Agent"})，它会逐块产出内容。chain.batch([{"topic": "Memory"}, {"topic": "Retrieval"}])则是一次传入多个输入并行处理。项目代码里这三种调用方式都可能出现，看到.invoke、.stream、.batch就明白对应的是单次、流式、批量场景。

2.2 Agent：让模型自己决定调用哪个工具，项目代码里长这样

Agent 要解决的问题是：模型不能只回答，还要“动手”。比如用户问“3 加 4 再乘以 2 等于多少”，模型如果直接口算，算错的风险很高，正确做法是调用加法工具和乘法工具。项目代码里的 agent_demo.py 通常会这样注册工具。

# agent_demo.py from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate from langchain_core.tools import tool @tool def add(a: int, b: int) -> int: """两个整数相加，结果返回整数。""" return a + b @tool def multiply(a: int, b: int) -> int: """两个整数相乘，结果返回整数。""" return a * b tools = [add, multiply] prompt = ChatPromptTemplate.from_messages([ ("system", "你是数学助手，必须用工具计算结果，不要自己推断。"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ]) # 注意：这里需要替换成支持工具调用的真实模型驱动类，FakeChatModel 只能演示 Chain agent = create_tool_calling_agent(fake_llm, tools, prompt) executor = AgentExecutor(agent=agent, tools=tools, verbose=True) resp = executor.invoke({"input": "3加4再乘以2等于多少？"}) print(resp)

先明确一点：上面这段是项目代码里的标准结构，但它能不能跑起来，取决于fake_llm是否允许工具调用。自定义的FakeChatModel只是返回固定文本，不具备任务决策能力，所以你在本地复现时要把这一行换成真实模型服务商提供的 ChatModel 驱动类，确认它支持 tool calling 再执行。

@tool装饰器做了两件事：一是把普通函数变成一个 Tool 对象，二是把函数签名和 docstring 转成模型能读到的 JSON Schema。模型的判断依据主要是函数名、参数类型和第一行描述。create_tool_calling_agent会把这些信息塞进系统上下文，AgentExecutor负责循环：模型先输出一个意图，执行器决定调用哪个工具，拿到结果后追加到agent_scratchpad，再交回给模型，直到模型认为任务完成或达到最大步数。

这里有几个参数值得关注。verbose=True会打印中间步骤，调试时能看到模型先写了什么、工具返回了什么，这是排查 Agent 行为最重要的开关。max_iterations可以限制循环上限，防止模型在一个问题上反复调用工具停不下来。handle_parsing_errors=True则会在模型输出格式不合法时把错误信息回传给模型，让它自己修正，代价是多一次模型调用。

如果项目代码里用的是create_react_agent，那是另一套 ReAct 范式：模型需要按照思考、动作、动作输入、观察四段格式输出，再由解析器拆分。两套机制不要混用，看到create_tool_calling_agent就别尝试用 ReAct 的 Prompt 模板，反之亦然。

2.3 Memory 与 Retrieval：项目代码里的对话记忆和外部知识查询

大模型本身不记上下文，Memory 和 Retrieval 就是它的“外挂记忆”。Memory 管的是对话历史，Retrieval 管的是外部文档。两者经常被混淆，但代码形态完全不同。

# memory_demo.py from langchain.memory import ConversationBufferMemory from langchain.chains import ConversationChain memory = ConversationBufferMemory(return_messages=True) conversation = ConversationChain(llm=fake_llm, memory=memory, verbose=True) print(conversation.predict(input="你好，我叫A同学")) print(conversation.predict(input="请记住我的名字")) print(conversation.predict(input="你记得我叫什么吗？"))

ConversationBufferMemory会把每一轮对话原样缓存，下次调用时把全部历史拼进 Prompt。return_messages=True表示以消息列表返回，而不是拼成一整段文本；这对 ChatModel 更友好，因为它天然支持多轮消息结构。ConversationChain是老项目代码里最常见的写法，新版 LangChain 更推荐你直接用RunnableWithMessageHistory配合一个历史存储对象。入门代码里出现ConversationChain不奇怪，但你要知道它在新版里已经被标为过时，新项目建议换掉。

Retrieval 的离线演示代码一般是这样：

# retrieval_demo.py from langchain_core.embeddings import DeterministicFakeEmbeddings from langchain_core.vectorstores import InMemoryVectorStore texts = [ "LangChain可以用链的方式编排大模型调用", "Memory组件让对话记住历史信息", "Retrieval组件把外部知识检索进提示词", ] embeddings = DeterministicFakeEmbeddings(size=8) vector_store = InMemoryVectorStore.from_texts(texts, embedding=embeddings) retriever = vector_store.as_retriever(search_kwargs={"k": 2}) hits = retriever.invoke("LangChain的核心组件有哪些？") for doc in hits: print(doc.page_content)

这个演示刻意用了假嵌入和内存向量库，目的是让你在没有任何外部服务的情况下跑通检索流程。DeterministicFakeEmbeddings生成固定的随机向量，每个文本都会被变成一个数组；InMemoryVectorStore把这些向量放在内存里做近似搜索，重启进程就丢失。真实项目里，这两行会换成真正的嵌入模型和持久化向量库。

search_kwargs={"k": 2}是检索参数，表示返回最相近的前 2 段。k 值太大容易引入噪声，k 值太小可能漏掉关键信息。项目代码里如果检索质量不好，先不是改模型，而是先检查文本分块是否合理、嵌入模型是否一致，紧接着去看 4.4 的那组避坑记录。

3. 在本地跑通项目代码：依赖版本、配置项与最小复现路径

看懂结构之后，下一步就是动手跑。这个阶段最容易出问题的是环境和配置。LangChain 的安装逻辑过去几年变过多次，把环境搭对了，后面踩坑数量能少一大半。

3.1 先建虚拟环境再装依赖：Python 版本与包管理器怎么选

拿到项目代码，第一件事不是pip install langchain，而是先建一个干净的虚拟环境。LangChain 依赖树比较复杂，不同项目对版本要求差异很大，直接用全局环境装容易把系统 Python 搞乱。我一般会用下面这组命令。

python3.11 -m venv .venv source .venv/bin/activate pip install --upgrade pip pip install "langchain>=0.1,<1.0" langchain-core langchain-community pip install -e .

虚拟环境命令是通用做法，python3.11指 Python 3.11 解释器版本。LangChain 在 Python 3.8 到 3.12 上基本都能跑，但建议选 3.10 或 3.11，这个区间对异步调用和类型注解的支持最稳，也少遇到某些依赖包编译报错。langchain>=0.1,<1.0是一个比较宽的范围，因为 LangChain 从 0.1 开始把组件拆分到独立包，老代码里常见的langchain.llms在新版里变成了langchain_community或langchain_core。

langchain-core是核心抽象层，包含 BaseChatModel、Prompt 模板、输出解析器这些与供应商无关的部分。langchain-community是社区集成层，里面放着文档加载器、向量库、各类模型驱动。如果项目代码里用到了文本分切，还需要单独安装配套包，常见做法是把它们写进项目的依赖声明文件，用一条pip install -e .把当前项目及其依赖一起装入当前虚拟环境。这种方式的好处是，项目代码的入口脚本可以直接用包内模块名导入，不会出现路径找不到的问题。

还要提醒一句，Windows 上激活虚拟环境的命令是.venv\Scripts\activate，macOS 和 Linux 才是source .venv/bin/activate。跑通一次之后，建议用pip freeze > requirements.lock把当前环境版本固定下来，下次复现或交给其他同学时能少很多“在我这是好的”的争议。

3.2 最小运行：先跑通一个不带外部依赖的脚本

环境就绪后，不要急着跑 Agent，那个脚本依赖真实模型服务。先跑只靠 FakeChatModel 就能完成的最小脚本，验证框架安装有没有问题。

python src/chain_demo.py

预期输出是一行以“这是离线模拟回答”开头的话，内容会包含你传入的 topic 的前几个字。只要这个脚本能正常输出，就说明 langchain-core、langchain-community 以及项目里的核心模块都装对了。接下来再依次跑 memory_demo.py 和 retrieval_demo.py，这两个同样不依赖真实模型服务。

项目代码通常会把入口脚本放在src/目录下，有的还会带tests/目录。最小运行顺序可以参考这张表：

如果 chain_demo 就报错，先看报错信息里有没有ModuleNotFoundError。一眼扫到某个langchain.xxx找不到，基本就是版本拆分问题，直接去 4.1 看解法。如果提示缺少依赖包，用pip list看当前环境装了哪些包，再对照项目的依赖清单补齐。不要凭感觉装最新版，很多入门代码锁定的是一两个月前的 API，装太新的版本反而会跑挂。

3.3 接入真实模型：环境变量、Temperature、MaxTokens 等必调参数

跑通离线脚本后，就该让代码连上真实模型服务了。常见做法是把 API Key 和模型名放在环境变量里，避免硬编码进源码。项目代码里通常会有这样一个配置模块。

# llm_config.py import os from dataclasses import dataclass @dataclass class LLMConfig: model_name: str = "default-chat-model" temperature: float = 0.7 max_tokens: int = 512 top_p: float = 1.0 api_key: str = "" def load_config_from_env() -> LLMConfig: return LLMConfig( model_name=os.getenv("MODEL_NAME", "default-chat-model"), temperature=float(os.getenv("TEMPERATURE", "0.7")), max_tokens=int(os.getenv("MAX_TOKENS", "512")), top_p=float(os.getenv("TOP_P", "1.0")), api_key=os.getenv("LLM_API_KEY", ""), )

这段代码只负责读取配置，真正创建模型时需要把LLMConfig传给项目接入的模型驱动类。不同供应商的驱动类参数名略有区别，但model_name、temperature、max_tokens、top_p这四类是通用参数。

项目代码里的.env.example文件通常会列出这些变量名，复制一份改成.env，再配合一个环境变量加载工具就能使用。没有加载工具时也可以直接在终端里用export设置，效果一样。

调参时我的一个经验是：不要同时动 temperature 和 top_p。先固定一个，再单独调另一个，一次只改一个变量，否则你很难判断是哪个参数导致输出突然变差。温度调参在不同模型上的表现差异很大，很多现象只能用体感解释，所以每次修改都记录当时的输出，不然会陷入调参玄学。

4. LangChain 项目代码的 5 个常见坑：现象、原因、排查路径

读 LangChain 项目代码最常见的情况是“版本看着对但跑不通”。很多坑不是代码逻辑问题，而是框架演进造成的兼容性问题。下面五条是按出现频率排的，每一条都按现象、原因、解决三部分展开，方便你照着排查。

4.1 一导入就报错：ModuleNotFoundError 背后的 LangChain 包拆分

现象：运行脚本时出现ModuleNotFoundError: No module named 'langchain.llms'或者langchain.vectorstores。代码明明没问题，导入就是过不了。

原因：LangChain 0.1 之后做了包拆分，原先集中在langchain主包里的模型、向量库、文档加载器被移到了langchain-community或langchain-core。旧版入门项目代码写的导入路径自然就失效了。

解决：先确认当前版本，再更新导入路径。

python -c "import langchain; print(langchain.__version__)"

如果版本是 0.1 及以上，把旧写法改成新写法：

# 旧写法：from langchain.llms import YourLLM # 新写法：from langchain_community.llms import YourLLM

这里YourLLM只是占位符，真实项目里会是某个具体模型驱动类的名字，你在项目代码里看到哪个类导入失败，就把包名改成langchain_community再试。还有一种情况是langchain.chains里的某些链被拆到langchain主包的特定子模块，改动相对少，核心思路是先看报错路径，再判断它属于抽象层、社区集成层还是应用层。不要图省事直接降级到老版本，那样会丢掉后续安全修复，而且其他依赖也容易被拖进版本地狱。

4.2 Agent 不调用工具，反而把工具描述当答案输出

现象：Agent 脚本能正常执行，但模型没有调用add或multiply，反而在回复里写出“根据工具描述，你可以使用加法函数”，或者直接自己心算一个结果给你。

原因：通常是三种情况叠加。第一，你接入的模型驱动类不支持工具调用，LangChain 只是把工具列表传给模型，模型没有相关能力去解析和决策；第二，工具函数的描述太含糊，模型不知道什么情况下该触发它；第三，系统提示词里写了“必须用工具”，但没有说明判断条件，模型在挣扎后选择自由发挥。

解决：先换一个支持工具调用的模型驱动类，然后在工具 docstring 第一行写明触发条件。

@tool def add(a: int, b: int) -> int: """仅当用户要求计算加法时调用，返回 a + b 的整数结果。""" return a + b

docstring 第一行会被模型当作工具的主要说明，所以不要写“这是一个加法函数”这种同义反复，要让模型知道“用户在什么场景下应该选你”。改完工具描述后再把verbose=True打开，观察中间步骤。如果模型输出里出现了Action Input但解析器报错，说明输出格式带有多余字符，先在工具描述的结尾补一句“参数必须是合法 JSON”，或者开启handle_parsing_errors=True。这类问题最怕的就是只改 prompt 不看执行日志，日志会告诉你到底是模型不调工具，还是调了但解析失败。

4.3 Memory 越聊越慢，最后直接报告上下文超限

现象：开启 Memory 后，前几轮对话正常，聊到第 5 轮以后首字延迟明显变长，再往后直接报context length exceeded或类似错误。

原因：ConversationBufferMemory会把所有历史消息原封不动地拼进 Prompt，token 数是线性增长的。模型上下文窗口有限，塞满了之后被迫报错。很多入门项目代码天生就带这个坑，因为示例里只聊两三轮，看不出问题，一旦真用起来立刻翻车。

解决：换成带摘要能力的 Memory，给历史设一个 token 上限。

from langchain.memory import ConversationSummaryBufferMemory memory = ConversationSummaryBufferMemory( llm=fake_llm, # 真实项目替换为能生成摘要的模型驱动类 max_token_limit=300, return_messages=True, )

ConversationSummaryBufferMemory的逻辑是：当原始历史超过max_token_limit时，它会用llm生成一段摘要，然后只保留最近几条原始消息，其余全部替换成摘要。这样无论聊多久，历史长度都能被压在可控范围内。代价是每次触发摘要都会多一次模型调用，所以max_token_limit不能设得太小，否则每聊几轮就要摘要一次，成本和延迟反而更高。一个相对保险的起点是 300 到 500，再根据你任务的上下文长度和预算调整。

如果你只是临时验证，不想引入摘要，可以用手动截断：直接操作memory.chat_memory.messages，只保留最后 N 条，但这会丢掉前面所有的语义，不是长期方案。项目代码里如果看到 BufferMemory 跑在长对话场景，基本可以直接判断是设计失误。

4.4 Retrieval 查出来的是无关内容：分块、Embedding 与 k 值

现象：检索脚本没有报错，返回的doc.page_content却跟用户问题完全没关系，或者命中了一个通用句子但没命中真正有用的细节。

原因：大部分情况是分块方式不对。整段文本过长时，一条 chunk 里塞了多个主题，生成向量后这些语义相互抵消，检索效果自然变差。也可能是因为构建向量库和查询时用了不同的 Embedding 模型，导致向量空间不一致。还有一种可能是 k 值设得不准，太小漏掉关键段落，太大把噪声捞进来。

解决：先看分块参数，再确认 Embedding 一致。

from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=200, chunk_overlap=50, ) chunks = splitter.split_text(long_text)

chunk_size=200是按字符数切分，不是 token 数。过大会稀释语义，过小会切断一句话。chunk_overlap=50表示相邻块之间保留 50 个字符的重叠，确保被切到边界的那句话在两个块里都保持完整。切完以后先打印两段结果，肉眼看语义是否完整，这是最笨也最有效的检查方式。

嵌入模型不一致的问题更隐蔽。项目代码里如果分了两步建索引和查询，要确保构建vector_store时用的embeddings实例和retriever里实际查询用的实例是同一个模型。哪怕都是同一个向量库文件，如果查询时换了新模型，同一个句子的向量也可能需要重新生成。排查方法是在两处各打印一条文本的向量维度，维度变了或者数值范围对不上，基本就能锁定。

4.5 异步并发时遇到事件循环报错：Future attached to a different loop

现象：在 Jupyter Notebook 或者一段脚本里调用ainvoke或异步 Agent 时，报错RuntimeError: Event loop is closed或Future attached to a different loop。同步调用一切正常，改成异步就翻车。

原因：LangChain 的异步接口依赖 asyncio 事件循环。Notebook 环境自带一个正在运行的事件循环，代码里如果又用asyncio.run()创建新循环，二者会互相争抢，导致 Future 被绑定到错误的循环上。这种情况几乎和项目代码无关，是调用方式的问题。

解决：把整个异步流程收敛到一个统一的 async 函数入口里，只在最后调用一次asyncio.run。

import asyncio async def main(): resp = await agent_executor.ainvoke({"input": "3加4再乘以2等于多少？"}) print(resp) return resp if __name__ == "__main__": asyncio.run(main())

不要在 async 函数内部再嵌套asyncio.run，也不要在一个函数里反复创建新的事件循环。如果需要并发处理多个输入，用await asyncio.gather(...)把多个任务包在一起，等它们全部完成后统一返回。这个坑常见于从同步示例改成异步批量调用时，改的时候注意把入口收口，避免每个请求都自成一套循环。

5. 从跑通到敢改：给项目代码加一层观测，再做一次端到端验证

跑通只是起点，真正敢改项目代码，需要先知道每一次模型调用和工具调用到底发生了什么。LangChain 提供了一套回调机制，比verbose=True更清晰，也更容易沉淀成自己团队的调试工具。

5.1 给 LangChain 项目代码加一个回调：每一步发生了什么都能看见

# trace_callback.py from langchain_core.callbacks import BaseCallbackHandler class TraceCallback(BaseCallbackHandler): def on_llm_start(self, serialized, prompts, **kwargs): print(f"[LLM 启动] prompts={prompts}") def on_llm_end(self, response, **kwargs): text = response.generations[0][0].text print(f"[LLM 结束] text={text[:100]}") def on_tool_start(self, serialized, input_str, **kwargs): print(f"[工具调用] name={serialized.get('name')} input={input_str}") def on_tool_end(self, output, **kwargs): print(f"[工具结束] output={output}")

使用方式简单直接：在invoke时把回调实例放到 config 里。

chain.invoke({"topic": "LangChain"}, config={"callbacks": [TraceCallback()]})

模型一启动，回调就会打印传入的 prompt；模型一结束，就会打印生成的前 100 字；工具被调用时，会打印工具名和输入参数。这套东西的价值在于，你可以把项目代码当黑匣子打开，每次改动 prompt 或者调 temperature 后，直接在日志里对比模型的决策变化。回调还能统计 token 用量和耗时，只要在on_llm_start里记一个时间戳，在on_llm_end里做差，就能得到单次模型调用的耗时。

5.2 用一段代码记录端到端耗时，验证性能是不是真的可用

给项目代码做功能验证之外，还要做性能验证。最简单的做法是在调用入口前后各打一个时间点。

import time start = time.perf_counter() resp = agent_executor.invoke({"input": "3加4再乘以2等于多少？"}) elapsed_ms = (time.perf_counter() - start) * 1000 print(f"端到端耗时：{elapsed_ms:.1f} ms") print(f"输出：{resp['output']}")

time.perf_counter()比直接用time.time()更精确，适合测量短耗时调用。记录到的总耗时包括模型调用、工具执行和 Agent 循环调度，如果时间长得不可接受，再拆开看是哪一步慢。工具函数如果本身是同步的，常见瓶颈还是模型调用次数，比如 Agent 在循环里多做了几次决策，每次决策都是一次完整的大模型请求。

给端到端验证列一个检查清单，能让项目代码从“能跑”升级到“能用”：

我最早带同学入门时，总觉得跑通 demo 就算学会。后来发现，每次改参数都要靠猜，因为代码里没有观测层。自从习惯给最小脚本加一个回调，把模型输入输出和耗时都打印出来，很多“玄学”就变成了可见的调用序列。改代码前先看一次 trace，改完再看一次，对比差异就能知道问题出在模型还是链路。希望这些细节能帮你在《LangChain入门与实践》这份项目代码上少走几段弯路，也希望帮到你真正把这套框架敢用起来。

本文还有配套的精品资源，点击获取