2025年信息化基础设施维护服务机构排行，信息化基础设施维护

实用指南：LangChain完整教程：六大组件+分层设计+完整项目实战

news/2025/12/6 10:52:41/文章来源:https://www.cnblogs.com/yangykaifa/p/19315060

LangChain深度教程：从入门到精通的完整构建指南

前言

今天我想和大家分享一下我的LangChain学习历程,大多数文章要么止步于基础，要么直接跳入高级应用，却忽视了从理解到实践的关键桥梁。这份文章我不仅思考了如何来向大家展示一个直观的LangChain结构,还尽量保证内容的层层递进,你可以根据自己的需要来进行学习,希望我的总结能够帮助到大家,如果有更好的改进建议,也请大家指教,帮助我完善内容,以帮助更多的求知者。

背景

2023-2024年，随着LLM技术的爆发式发展，无数开发者和企业涌入这一领域。
2025年，大语言模型（LLM）已深度融入企业级应用开发，成为AI落地的关键技术。却很快发现了一个根本性矛盾：LLM本质上是基于过去数据的"经验总结者"，它无法获取实时信息，更无法访问企业私有知识库。当业务场景要求AI回答"最新政策解读"或"内部项目进度"时，传统LLM应用往往陷入"知识断层"的困境。
根据2025年Gartner最新报告，83%的AI应用失败源于"数据孤岛"问题，而非模型能力不足。LangChain正是为解决这一行业痛点而生——它不是简单的API封装，而是一套完整的开发范式，让开发者能够构建真正具备商业价值的LLM应用。

一、什么是LangChain?

LangChain是一个开源的LLM应用开发框架，它通过组件化、可组合的架构，让开发者能够高效构建端到端的LLM驱动应用。简单来说，LangChain解决了“如何让大语言模型（LLM）真正理解并操作你的私有数据和业务系统”这一核心问题。

官方定义：

LangChain是一个用于开发由语言模型驱动的应用程序的框架。它使应用程序能够：
具有上下文感知：连接语言模型到其他数据源
具有推理能力：允许语言模型与其环境交互

1. LangChain vs 传统LLM应用

维度	传统LLM应用	LangChain应用
数据访问	仅限于训练数据	可连接任意数据源（数据库、API、文档等）
交互能力	单次问答，无状态	多轮对话，有记忆和上下文
任务复杂度	简单问答、文本生成	复杂工作流、工具调用、自主决策
开发模式	点对点API调用	组件化、管道式开发
可扩展性	有限，需要重写逻辑	模块化，易于扩展和迭代

轻松理解：
传统LLM应用像是“孤立的专家”，只能基于已有知识回答问题；而LangChain应用则是“连接的团队”，能够调用各种工具和资源来解决实际问题。

2. LangChain如何解决数据孤岛

LangChain通过以下方式解决这一问题：

数据感知： 将语言模型连接到企业数据源（数据库、文档、API等）
自主性： 允许语言模型与外部环境交互（查询数据库、调用API、执行计算等）

3. LangChain的核心价值

LangChain的价值不在于它提供了多少功能，而在于它为LLM应用开发提供了一个系统化的框架，使开发者能够：

组件化： 将复杂应用拆解为可复用的模块（模型、提示词、检索器、工具等）
可组合： 轻松组合不同组件，构建复杂工作流
可扩展： 随着业务需求变化，灵活调整应用结构

就类似于组装积木一样,你可以根据需要灵活的区拼装你想要的形状。

二、学习LangChain需要哪些前置知识?

学习LangChain并不需要你是Python或AI领域的专家，但需要掌握一些关键基础。相信我,这些前置知识将帮助你更快地理解LangChain的原理和应用，避免在入门阶段陷入不必要的困惑。而且不仅仅是在学习LangChain的阶段,学习下面的部分内容也有助于提升你的工程化思维和能力。

1. Python基础：

LangChain是用Python编写的开源框架，因此掌握Python基础是必不可少的。

2. LLM基础：理解大语言模型的本质

LangChain是围绕LLM构建的框架，因此理解LLM的基本概念至关重要。

3. 工具链基础：了解现代AI开发环境

LangChain作为连接LLM与外部系统的桥梁，需要了解以下工具链:API调用、向量数据库、环境管理。

三、LangChain的整体组成架构

LangChain的架构设计遵循 “组件化、可组合、可扩展” 的原则，将复杂的LLM应用开发拆解为多个相互协作的模块。这种设计不仅使开发过程更加清晰，也大大提高了应用的可维护性和扩展性。

1. 分层架构 vs 核心组件

大家或许经常能够看到LangChain的六大核心组件讲解,却没提到LangChain的分层架构设计。分层架构和核心组件是两个不同维度的概念，它们描述的是LangChain的不同方面：
分层架构：描述的是LangChain框架的组织结构，即如何将框架内部功能按逻辑层次划分。
核心组件：描述的是LangChain框架提供的主要功能模块，即框架的"积木块"。

2. 分层架构（四个层次）

官方架构:
在这里插入图片描述

2.1 核心层(langchain-core)：

定位：框架的基石，定义基础抽象和通用接口

核心层是LangChain的"宪法"，它不包含具体实现，而是定义了整个框架必须遵守的规则和标准。

核心职责：

基础抽象定义：提供Runnable、BaseMessage、BaseRetriever等核心接口
可组合运行时：实现LCEL（LangChain Expression Language），让组件可以像管道一样连接
序列化协议：统一配置和状态的序列化标准
事件系统：提供组件间的通信机制

2.2 集成层 (Community Integrations)：

定位：连接外部世界的桥梁，提供丰富的第三方集成

集成层是LangChain生态最丰富的部分，包含了与各种外部服务和工具的连接器。

主要集成类别：

模型提供商：OpenAI、Anthropic、Google、Meta、智谱AI、通义千问等
向量数据库：Chroma、Pinecone、Weaviate、Milvus、Qdrant等
文档加载器：PDF、Word、Excel、网页、数据库、API等
工具集成：计算器、搜索引擎、API调用、代码执行等

代码结构：

# 集成层的典型使用
from langchain_community.llms import OpenAI
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import PyPDFLoader
from langchain_community.tools import DuckDuckGoSearchRun
# 各种第三方集成
llm = OpenAI(api_key="...")
vectorstore = Chroma.from_documents(documents, embeddings)
loader = PyPDFLoader("document.pdf")
search_tool = DuckDuckGoSearchRun()

2.3 应用层(langchain)：

定位：预构建的高级组件和解决方案模板

应用层基于核心层抽象和集成层能力，提供了开箱即用的高级功能，这是大多数开发者直接接触的部分。

核心组件：

链：预定义的工作流程，如RetrievalQA、ConversationalRetrievalChain
代理：具备推理和工具使用能力的智能体
检索器：文档检索和相似度搜索的高级封装
记忆：对话历史管理和上下文维护

典型使用：

from langchain.chains import RetrievalQA
from langchain.agents import initialize_agent, AgentType
# 应用层提供的预构建解决方案
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
return_source_documents=True
)
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)

2.4 编排层(langgraph)：

定位：复杂工作流和状态管理的终极解决方案

编排层是LangChain体系中最先进的部分，专门处理需要状态维护、循环、分支等复杂逻辑的应用场景。

核心概念：

状态图：基于状态机的工作流定义
持久化状态：支持长时间运行的复杂任务
多代理协作：多个AI代理协同工作的框架
循环和条件：支持if-else、for、while等控制流

典型应用场景：

from langgraph.graph import StateGraph, END
from typing import Dict, Any
# 定义状态
class AgentState(TypedDict):
messages: List[BaseMessage]
next: str
# 构建状态图
graph = StateGraph(AgentState)
# 添加节点和边
graph.add_node("agent", call_model)
graph.add_node("tools", execute_tools)
graph.add_edge("tools", "agent")
graph.set_entry_point("agent")
# 编译为可执行图
app = graph.compile()

3. 核心组件（6个）

LangChian六大组件
此图片来源：机器学习社区

3.1 LangChain Core (基础层)

定位：所有组件的通用基础，提供统一的接口标准和运行时环境

LangChain Core定义了所有组件都必须遵守的基本规则和通信协议。

核心功能：

可运行接口：Runnable协议，让所有组件可以像函数一样被调用和组合
消息系统：标准化AI交互的消息格式（HumanMessage、AIMessage、SystemMessage）
序列化：统一的配置保存和加载机制
LCEL语言：LangChain表达式语言，提供声明式的组件组合方式

技术示例：

from langchain_core.runnables import RunnableLambda
from langchain_core.messages import HumanMessage
# 任何对象都可以通过Runnable接口接入LangChain
def double_input(x):
return x * 2
runnable = RunnableLambda(double_input)
result = runnable.invoke(2)  # 返回4
# 标准化的消息格式
messages = [
SystemMessage(content="你是一个助手"),
HumanMessage(content="你好！")
]

设计价值：Core层确保了整个生态的互操作性，让开发者可以自由组合不同来源的组件。

3.2 Model I/O (模型输入输出)

定位：与LLM交互的标准接口，统一不同模型的调用方式

Model I/O组件解决了"如何与各种LLM对话"的问题，无论模型来自哪个提供商，都提供一致的编程接口。

核心模块：

模型：ChatModels、LLMs - 统一的模型调用接口
提示词：PromptTemplates - 动态提示词管理和格式化
输出解析：OutputParsers - 结构化输出处理

完整工作流示例：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
# 1. 模型初始化
llm = ChatOpenAI(model="glm-4-flash", temperature=0.7)
# 2. 提示词模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个{role}，用{style}风格回答"),
("human", "{question}")
])
# 3. 输出解析器
parser = JsonOutputParser()  # 或 StrOutputParser()
# 构建完整管道
chain = prompt | llm | parser
# 执行
result = chain.invoke({
"role": "技术专家",
"style": "简洁明了",
"question": "解释什么是RAG"
})

设计价值：Model I/O让开发者无需关心底层模型差异，可以专注于业务逻辑。

3.3 Retrieval (检索)

定位：让LLM能够访问和利用外部知识体系

检索组件是LangChain解决"数据孤岛"问题的核心技术，通过RAG模式让LLM具备访问私有知识的能力。

核心流程：

文档加载 → 文本分割 → 向量化 → 存储 → 检索 → 增强生成

完整实现示例：

from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
# 1. 文档加载
loader = WebBaseLoader("https://example.com/article")
documents = loader.load()
# 2. 文本分割
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
chunks = text_splitter.split_documents(documents)
# 3. 向量化存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)
# 4. 检索增强生成
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever
)
# 使用
answer = qa_chain.run("文章的主要观点是什么？")

高级特性：

多路检索：结合语义检索和关键词检索
重排序：对检索结果进行智能排序
混合搜索：同时支持向量搜索和元数据过滤

3.4 Agents (智能体)

定位：具备推理和决策能力的自主AI系统

Agents是LangChain最强大的组件，让LLM从被动回答升级为主动的决策和执行，Agent具备推理能力，能够自主分析问题、制定计划、选择工具并执行任务。

Agent基于ReAct（Reasoning + Acting）模式工作：
在这里插入图片描述

核心概念：

推理循环：思考(Reason) → 行动(Act) → 观察(Observe) → 再思考(Reason)
工具使用：根据任务选择并调用合适的工具
自主决策：决定何时完成任务或继续探索

典型Agent示例：

from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType
# 定义工具
tools = [
Tool(
name="搜索",
func=search_tool.run,
description="用于搜索最新信息"
),
Tool(
name="计算器",
func=calculator.run,
description="用于数学计算"
)
]
# 初始化Agent
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
handle_parsing_errors=True
)
# 执行复杂任务
result = agent.run(
"找出特斯拉当前股价，计算如果投资10000美元可以买多少股"
)

Agent类型：

ReAct Agent：最常用的推理+行动模式
Plan-and-Execute：先制定计划再执行
Self-Ask：通过自问自答分解复杂问题

3.5 Tools (工具)

定位：扩展LLM能力边界的外部功能接口

Tools让LLM具备了"动手能力"，可以调用各种外部服务和API来完成自身无法完成的任务。

工具分类：

基础工具：计算器、时钟、文件操作
网络工具：搜索引擎、API调用、网页抓取
专业工具：数据库查询、代码执行、数据分析
自定义工具：根据业务需求定制

工具定义和使用：

from langchain.agents import tool
from datetime import datetime
# 自定义工具
@tool
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
"""获取指定时区的当前时间"""
now = datetime.now()
return f"当前时间：{now.strftime('%Y-%m-%d %H:%M:%S')}"
@tool
def query_database(query: str) -> str:
"""执行数据库查询"""
# 实际的数据查询逻辑
return f"查询结果：{query}"
# 工具注册
tools = [get_current_time, query_database]
# 在Agent中使用
agent = initialize_agent(tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION)

设计价值：Tools系统让LangChain具备了无限扩展性，可以集成任何外部系统。

3.6 Memory (记忆)

定位：维护对话上下文和历史，实现有状态的连续交互

Memory组件让LLM应用具备了"记忆能力"，能够在多轮对话中保持上下文连贯性。

记忆类型：

对话缓冲：保存最近的对话记录
对话摘要：对历史对话进行摘要
向量存储：基于语义的长期记忆
自定义记忆：特定场景的记忆逻辑

记忆系统实现：

from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
# 1. 缓冲记忆 - 保存完整对话历史
buffer_memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
# 2. 摘要记忆 - 对长对话进行摘要
summary_memory = ConversationSummaryMemory.from_llm(
llm=llm,
memory_key="summary_history"
)
# 在链中使用记忆
from langchain.chains import ConversationChain
conversation = ConversationChain(
llm=llm,
memory=buffer_memory,
verbose=True
)
# 多轮对话
conversation.predict(input="你好，我是小明")
conversation.predict(input="你还记得我的名字吗？")  # 能记住上下文

高级记忆模式：

# 结合向量数据库的长期记忆
from langchain.vectorstores import Chroma
from langchain.memory import VectorStoreRetrieverMemory
retriever = vectorstore.as_retriever(search_kwargs={"k": 1})
memory = VectorStoreRetrieverMemory(retriever=retriever)
# 记忆会自动保存和检索相关上下文
memory.save_context({"input": "我喜欢Python"}, {"output": "好的，已记录"})
memory.load_memory_variables({"input": "我喜欢的编程语言"})

组件协作：完整应用示例

这六个组件如何协同工作？看一个完整的客户服务Agent示例：

from langchain.agents import AgentExecutor, create_react_agent
from langchain import hub
# 1. Model I/O - 模型和提示词
llm = ChatOpenAI(model="glm-4-flash")
prompt = hub.pull("hwchase17/react")
# 2. Tools - 业务工具
tools = [product_search, order_lookup, customer_info]
# 3. Memory - 对话记忆  
memory = ConversationBufferMemory(memory_key="chat_history")
# 4. Agent - 智能决策
agent = create_react_agent(llm, tools, prompt)
# 5. Executor - 执行引擎
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True
)
# 运行：所有组件协同工作
result = agent_executor.invoke({
"input": "帮我查找订单123的状态，然后告诉客户最新情况"
})

核心组件价值总结

这六个组件形成了一个完整的LLM应用开发生态：

Core 提供基础规则
Model I/O 处理模型交互
Retrieval 扩展知识边界
Agents 实现智能决策
Tools 连接外部世界
Memory 维护交互状态

在这里插入图片描述

四、环境搭建：LangChain环境配置

1. Python环境要求

LangChain要求Python 3.8或更高版本。建议使用Python 3.9或3.10，这些版本在兼容性和稳定性方面表现最佳。

2. 创建虚拟环境

强烈建议使用conda或venv创建虚拟环境，避免包冲突和依赖问题。
方法一：使用Conda

# 创建名为langchain-env的虚拟环境，指定Python版本
conda create -n langchain-env python=3.10
# 激活虚拟环境
conda activate langchain-env

方法二：使用venv（系统自带）

# 创建虚拟环境
python -m venv langchain-env
# 激活虚拟环境
# Windows:
langchain-env\Scripts\activate
# macOS/Linux:
source langchain-env/bin/activate

3 创建项目结构:

 LangChain入门/
│    config/                    # 配置文件目录
│   │    settings.py            # 项目配置和设置
│   │
│    data/                      # 数据文件目录
│   │    Lifestyle_and_Health_Risk_Prediction_Synthetic_Dataset.csv
│   │
│    tools/                     # 工具类目录
│   │    search_tools.py        # 搜索引擎工具实现
│   │
│    tests/                     # 测试文件目录（预留）
│   │
│    .env                       # 环境变量配置文件
│    app.py                     # 基础对话版本（教学示例1）
│    agent_demo.py              # Agent版本（教学示例2）
│    llm_zhipu.py               # 智谱AI LLM 包装器
│    custom_parser.py           # 自定义输出解析器
│    setup_and_test.py          # 环境验证和测试脚本
│    requirements.txt           # 项目依赖列表

4. 安装LangChain及相关依赖

创建requirements.txt文件：

# LangChain
langchain==0.3.27
langchain-community==0.3.31
langchain-core==0.3.78
langchain-openai==0.3.9
# 国内搜索引擎依赖
requests>=2.31.0
beautifulsoup4>=4.12.0
lxml>=4.9.0
# 环境管理
python-dotenv>=1.0.0

安装依赖：

# 使用pip安装所有依赖
pip install -r requirements.txt
# 或者逐个安装核心包
pip install langchain langchain-community langchain-core langchain-openai
pip install python-dotenv chromadb tiktoken

5. 获取API密钥（免费模型）

我们使用腾讯元宝、智谱AI等免费的AI模型,方便大家的练习。

腾讯元宝 (Yuanbao) - 备选方案:

6. 配置环境变量

在项目根目录创建.env文件：

# 智谱AI配置 (推荐)
ZHIPU_API_KEY=你的智谱AI_API_Key

# 腾讯元宝配置 (备选)
TENCENT_API_KEY=你的腾讯API_Key

五、实战入门：从基础运用到知识库问答

完成了以上的学习和准备工作,我们接下来就可以开始进行实操。

1. 创建第一个LangChain应用

1.1 创建配置模块

文件：config/settings.py
作用：集中管理所有配置参数

import os
from dotenv import load_dotenv
from typing import Optional
# 加载环境变量
load_dotenv()
# 智谱AI配置
ZHIPU_API_KEY = os.getenv("ZHIPU_API_KEY")
ZHIPU_API_BASE = "https://open.bigmodel.cn/api/paas/v4"
# 模型配置
MODEL_NAME = "glm-4-flash"
MODEL_TEMPERATURE = 0.7
MODEL_MAX_TOKENS = 2048
# 搜索配置
SEARCH_RESULT_LIMIT = int(os.getenv("SEARCH_RESULT_LIMIT", 5))
SEARCH_TIMEOUT = int(os.getenv("SEARCH_TIMEOUT", 30))
# 应用配置
DEBUG = os.getenv("DEBUG", "False").lower() == "true"
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
# 记忆配置
MEMORY_WINDOW_SIZE = 10
def validate_config():
"""验证配置是否完整"""
if not ZHIPU_API_KEY:
raise ValueError("ZHIPU_API_KEY 环境变量未设置")
return True

1.2 创建自定义LLM包装器

文件：llm_zhipu.py
作用：封装智谱AI接口，提供标准LangChain接口

import os
import logging
from typing import Any, Dict, List, Optional
from langchain.schema import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain.llms.base import LLM
from zhipuai import ZhipuAI
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ZhipuAILLM(LLM):
"""包装器"""
model_name: str = "glm-4-flash"
temperature: float = 0.7
max_tokens: int = 2048
api_key: str = ""
class Config:
"""Pydantic配置"""
arbitrary_types_allowed = True
extra = "ignore"
def __init__(self, **data: Any):
super().__init__(**data)
@property
def _llm_type(self) -> str:
return "zhipuai"
def _call(self, prompt: str, stop: Optional[List[str]] = None, **kwargs: Any) -> str:
"""调用智谱AI API"""
# 获取API密钥
api_key = self.api_key or os.getenv("ZHIPU_API_KEY")
if not api_key:
error_msg = "ZHIPU_API_KEY环境变量未设置"
logger.error(error_msg)
return error_msg
try:
client = ZhipuAI(api_key=api_key)
logger.info(f"调用智谱AI API，模型: {self.model_name}, 输入长度: {len(prompt)}")
response = client.chat.completions.create(
model=self.model_name,
messages=[{"role": "user", "content": prompt}],
temperature=self.temperature,
max_tokens=self.max_tokens,
)
result = response.choices[0].message.content
logger.info(f" API调用成功，输出长度: {len(result)}")
return result
except Exception as e:
error_msg = f" API调用失败: {str(e)}"
logger.error(error_msg)
return error_msg
def _convert_message_to_dict(self, message: BaseMessage) -> Dict[str, Any]:
"""转换消息"""
if isinstance(message, HumanMessage):
return {"role": "user", "content": message.content}
elif isinstance(message, AIMessage):
return {"role": "assistant", "content": message.content}
elif isinstance(message, SystemMessage):
return {"role": "system", "content": message.content}
else:
return {"role": "user", "content": str(message.content)}
class ZhipuAIChatLLM(ZhipuAILLM):
"""智谱AI LLM"""
def generate_chat(self, messages: List[BaseMessage], **kwargs: Any) -> str:
"""生成聊天响应"""
# 获取API密钥
api_key = self.api_key or os.getenv("ZHIPU_API_KEY")
if not api_key:
error_msg = "ZHIPU_API_KEY环境变量未设置"
logger.error(error_msg)
return error_msg
try:
# 每次调用都创建新的客户端
client = ZhipuAI(api_key=api_key)
formatted_messages = [self._convert_message_to_dict(msg) for msg in messages]
logger.info(f"消息数量: {len(formatted_messages)}")
response = client.chat.completions.create(
model=self.model_name,
messages=formatted_messages,
temperature=self.temperature,
max_tokens=self.max_tokens,
)
result = response.choices[0].message.content
logger.info(f"API调用成功，输出长度: {len(result)}")
return result
except Exception as e:
error_msg = f"出错: {str(e)}"
logger.error(error_msg)
return error_msg

1.3 创建基础对话版本

编写app.py
作用：演示基础的LLMChain + Memory使用

import os
from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory
from langchain.chains import LLMChain
# 导入我们自定义的智谱AI LLM
from llm_zhipu import ZhipuAIChatLLM
# 加载环境变量
load_dotenv()
# 初始化LLM
llm = ZhipuAIChatLLM(
model_name="glm-4-flash",
temperature=0.7,
max_tokens=2048,
api_key=os.getenv("ZHIPU_API_KEY")
)
# 创建记忆 - 保存对话历史
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
# 创建提示模板 - 包含对话历史
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个友好的个人助手，回答要自然、有帮助。"),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}")
])
# 构建链
chain = LLMChain(
llm=llm,
prompt=prompt,
memory=memory,
verbose=True
)
def chat_with_assistant():
print("个人助手已启动! 输入'退出'或'quit'结束对话")
print("-" * 50)
while True:
user_input = input("\n你: ").strip()
if user_input.lower() in ['退出', 'quit', 'exit']:
print("助手: 再见！很高兴为您服务！")
break
if not user_input:
continue
try:
# 调用链并获取响应
response = chain.invoke({"input": user_input})
print(f"助手: {response['text']}")
except Exception as e:
print(f"出错: {e}")
if __name__ == "__main__":
chat_with_assistant()

1.4 运行应用

python app.py

1.5 结果展示

在这里插入图片描述

2. 进阶应用：构建你的第一个Agent

2.1 创建搜索工具

文件：tools/search_tools.py
作用：提供网络搜索能力

import requests
from langchain.tools import Tool
from typing import Optional, Dict, Any
import json
import time
import logging
from bs4 import BeautifulSoup
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class BaiduSearchTool:
"""百度搜索"""
def __init__(self):
self.base_url = "https://www.baidu.com/s"
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2',
}
def search(self, query: str, num_results: int = 5) -> str:
"""
使用百度网页搜索（带重试机制）
"""
for attempt in range(3):
try:
params = {
'wd': query,
'rn': num_results,
'ie': 'utf-8',
'oq': query
}
response = requests.get(
self.base_url,
params=params,
headers=self.headers,
timeout=30
)
if response.status_code == 200:
result = self._parse_baidu_results_robust(response.text, query, num_results)
if "未找到" not in result and "失败" not in result:
return result
else:
logger.warning(f"百度搜索结果为空，尝试第 {attempt + 1} 次重试")
else:
logger.warning(f"百度搜索HTTP错误 {response.status_code}，尝试第 {attempt + 1} 次重试")
except Exception as e:
logger.warning(f"百度搜索异常 {str(e)}，尝试第 {attempt + 1} 次重试")
if attempt < 2:
time.sleep(1)
return self._get_fallback_result(query)
def _parse_baidu_results_robust(self, html: str, query: str, num_results: int) -> str:
"""百度搜索结果解析"""
try:
soup = BeautifulSoup(html, 'html.parser')
results = []
# 尝试多种可能的百度搜索结果容器选择器
possible_selectors = [
'div.result',
'div.c-container',
'div[class*="result"]',
'div[class*="c-container"]',
'div.contentLeft',
]
containers = []
for selector in possible_selectors:
containers.extend(soup.select(selector))
# 如果上述选择器都没找到，尝试查找包含标题和摘要的通用结构
if not containers:
# 查找所有包含链接和文本的容器
link_containers = soup.find_all(['div', 'table'], class_=lambda x: x and any(keyword in str(x) for keyword in ['result', 'c-', 'content']))
containers = link_containers
for i, container in enumerate(containers[:num_results]):
try:
# 多种标题提取策略
title = None
title_selectors = ['h3', 'h3 a', 'a[data-click]', '.t a', '.c-title a']
for selector in title_selectors:
title_elem = container.select_one(selector)
if title_elem:
title = title_elem.get_text().strip()
break
if not title:
# 尝试在容器内查找第一个链接
link_elem = container.find('a')
title = link_elem.get_text().strip() if link_elem else "无标题"
# 多种摘要提取策略
content = None
content_selectors = ['.c-abstract', '.c-span-last', '.c-gap', '.content-right_8Zs40']
for selector in content_selectors:
content_elem = container.select_one(selector)
if content_elem:
content = content_elem.get_text().strip()
break
if not content:
# 尝试查找包含描述文本的元素
desc_elements = container.find_all(['div', 'span'], class_=lambda x: x and any(keyword in str(x) for keyword in ['abstract', 'content', 'desc']))
if desc_elements:
content = desc_elements[0].get_text().strip()
else:
# 提取容器中除标题外的文本
if title_elem:
title_elem.extract()
content = container.get_text().strip()[:100] + "..." if container.get_text().strip() else "无内容摘要"
results.append(f"{i+1}. {title}\n   {content}")
except Exception as e:
logger.debug(f"解析单个搜索结果时出错: {str(e)}")
continue
if results:
return f"百度搜索 '{query}' 的结果:\n" + "\n\n".join(results)
else:
return f"未找到关于 '{query}' 的搜索结果，请尝试其他关键词"
except Exception as e:
logger.error(f"解析百度搜索结果时出错: {str(e)}")
return f"解析百度搜索结果时出错: {str(e)}"
def _get_fallback_result(self, query: str) -> str:
"""备用搜索结果"""
return f"""搜索 '{query}' 的结果（网络搜索暂时不可用）:
建议您：
1. 检查网络连接
2. 尝试使用更具体的关键词
3. 稍后重试
当前提供的信息基于模型训练数据，可能不是最新的。"""
class ReliableSearchTool:
"""搜索工具"""
def __init__(self):
self.baidu_tool = BaiduSearchTool()
def search(self, query: str, num_results: int = 5) -> str:
"""
使用百度搜索
"""
return self.baidu_tool.search(query, num_results)
def create_search_tools():
"""搜索工具集合"""
reliable_search = ReliableSearchTool()
tools = [
Tool(
name="web_search",
func=reliable_search.search,
description="使用网络搜索引擎获取最新信息。输入：搜索查询字符串"
),
Tool(
name="baidu_search",
func=reliable_search.search,
description="使用百度获取内容。输入：搜索查询字符串"
)
]
return tools
# 创建工具实例
search_tools = create_search_tools()

2.2 创建自定义输出解析器

文件：custom_parser.py
作用：处理Agent的输出格式

import re
import logging
from typing import Union
from langchain.agents import AgentOutputParser
from langchain.schema import AgentAction, AgentFinish
logger = logging.getLogger(__name__)
class CustomOutputParser(AgentOutputParser):
"""解析器，处理智谱AI的输出格式"""
def parse(self, text: str) -> Union[AgentAction, AgentFinish]:
logger.info(f"解析Agent输出: {text}")
# 检查是否包含Final Answer
if "Final Answer:" in text:
# 提取最终答案
final_answer_match = re.search(r"Final Answer:\s*(.*)", text, re.DOTALL)
if final_answer_match:
answer = final_answer_match.group(1).strip()
logger.info(f"提取到最终答案: {answer}")
return AgentFinish(
return_values={"output": answer},
log=text,
)
# 检查是否包含Action和Action Input
action_match = re.search(r"Action:\s*(.*?)\nAction Input:\s*(.*)", text, re.DOTALL)
if action_match:
action = action_match.group(1).strip()
action_input = action_match.group(2).strip().strip('"')
logger.info(f"提取到Action: {action}, Input: {action_input}")
return AgentAction(tool=action, tool_input=action_input, log=text)
# 如果无法解析，返回最终答案
logger.warning(f"无法解析Agent输出，直接返回文本: {text}")
return AgentFinish(
return_values={"output": text},
log=text,
)

2.3 创建Agent版本

文件：agent_demo.py
作用：演示完整的Agent + Tools使用

import os
import logging
from dotenv import load_dotenv
from langchain.agents import initialize_agent, AgentType
from langchain.memory import ConversationBufferWindowMemory
from langchain.prompts import MessagesPlaceholder
from langchain.schema import SystemMessage
from llm_zhipu import ZhipuAIChatLLM
from tools.search_tools import search_tools
from config.settings import validate_config
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 加载环境变量并验证配置
load_dotenv()
try:
validate_config()
except Exception as e:
logger.error(f"配置验证失败: {e}")
raise
class HealthcareAgent:
def __init__(self):
# 使用智谱AI LLM
self.llm = ZhipuAIChatLLM(
model_name="glm-4-flash",
temperature=0.1,  # 降低温度
max_tokens=2048,
api_key=os.getenv("ZHIPU_API_KEY")
)
# 创建记忆
self.memory = ConversationBufferWindowMemory(
memory_key="chat_history",
k=10,
return_messages=True
)
# 系统提示词
system_message = SystemMessage(content="""
你是一个专业的健康和生活方式助手，专门帮助用户分析健康风险并提供建议。
你的能力包括：
1. 搜索最新的健康信息和医学知识
2. 分析生活方式对健康的影响
3. 提供个性化的健康建议
4. 回答关于疾病预防和健康管理的问题
请遵循以下原则：
- 提供准确、科学的健康信息
- 对于严重的健康问题，建议用户咨询专业医生
- 尊重用户隐私，不询问过于私密的信息
- 用友好、专业的方式与用户交流
当需要最新信息时，请使用搜索工具。
重要：对于简单的问题（如日期、时间等），请直接回答，不要过度使用搜索工具。
""")
try:
# 初始化Agent
self.agent = initialize_agent(
tools=search_tools,
llm=self.llm,
agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,  # 对话型Agent
verbose=True,
memory=self.memory,
agent_kwargs={
"system_message": system_message,
"memory_prompts": [MessagesPlaceholder(variable_name="chat_history")],
"input_variables": ["input", "chat_history", "agent_scratchpad"]
},
handle_parsing_errors=True,
max_iterations=3,  # 限制最大迭代次数，避免无限循环
early_stopping_method="generate"  # 提前停止
)
logger.info("HealthcareAgent初始化成功")
except Exception as e:
logger.error(f"HealthcareAgent初始化失败: {e}")
raise
def chat(self, user_input: str) -> str:
"""与Agent对话"""
try:
# 对于简单问题，直接使用LLM回答
simple_questions = ["今天的日期", "现在几点", "当前时间", "日期", "时间"]
if any(q in user_input for q in simple_questions):
logger.info("检测到简单问题，直接使用LLM回答")
from datetime import datetime
now = datetime.now()
if "日期" in user_input:
return f"今天是{now.strftime('%Y年%m月%d日')}"
elif "时间" in user_input or "几点" in user_input:
return f"现在是{now.strftime('%H点%M分')}"
response = self.agent.run(user_input)
return response
except Exception as e:
logger.error(f"Agent对话过程中出错: {e}")
return f"抱歉，处理您的请求时出现了错误: {str(e)}"
def main():
"""主函数 - 测试Agent"""
print("=" * 60)
print("          健康与生活方式助手 Agent")
print("=" * 60)
print("注意：本助手提供的信息仅供参考，不能替代专业医疗建议")
print("输入 '退出' 或 'quit' 结束对话")
print("-" * 60)
try:
# 创建Agent实例
agent = HealthcareAgent()
while True:
try:
user_input = input("\n您: ").strip()
if user_input.lower() in ['退出', 'quit', 'exit']:
print("\n助手: 感谢使用！祝您健康！")
break
if not user_input:
continue
print("\n助手: ", end="", flush=True)
response = agent.chat(user_input)
print(response)
except KeyboardInterrupt:
print("\n\n助手: 对话已中断，再见！")
break
except Exception as e:
print(f"\n助手: 抱歉，出现了错误: {str(e)}")
except Exception as e:
print(f"初始化Agent失败: {e}")
print("请检查：")
print("1. ZHIPU_API_KEY环境变量是否正确设置")
print("2. 网络连接是否正常")
print("3. 依赖包是否正确安装")
if __name__ == "__main__":
main()

2.4 测试功能

我们进行简单的对话，测试一下联网搜索功能：
在这里插入图片描述

在这里插入图片描述

3. 完整项目链接

我已将项目上传至Git Code，链接如下
点击访问完整项目
项目链接：https://gitcode.com/2301_77705369/Langchain_introduction

七、未来展望

站在2025年的技术拐点，我们一路见证AI应用从"玩具"到"工具"再到"伙伴"的演进历程。学习LangChain不仅仅是掌握一个技术框架，更是拥抱一种新的编程范式。在这个快速发展的领域，保持好奇心、实践精神和开放心态比任何具体技术都更加重要。

技术转变

主动性

AI的能力不再局限于被动响应，而是朝着主动识别潜在需求和问题的方向发展。它可能会在工作中，提醒你是否需要总结最近更新的资料，或者自动优化你的工作安排，就像一位时刻保持清醒的助手

长期性

随着单次对话的记忆能力不断进步，以后的AI能扩展到持续的项目协作和目标追踪。哪怕你在一个月之后提起你都遗忘大部分细节的项目，它都能在你所疏漏的地方告诉你，在你需要的地方提起，而不是全篇大论的从头讲起。

情感性

随着AI对于情感和语言的理解不断加深，或许会出现与工作者性格互补的个性化伙伴。它能在你疲惫的时候调整工作，调整对话方式，在保证工作正常推进的情况下，同时兼顾你的身体健康和心理情绪。

对于我们而言

引导式开发

新的引导式开发真正逐渐的显现，不再是传统的具体编写逻辑性代码，而是通过目标来引导和约束AI进行输出，我们则作为最后的验证者，去不断优化。

技术栈分层

随着技术成熟，将出现更加专业化的技术栈分层
从基础模型层到编排框架层，再到领域应用层，最后到治理监控层
在这里插入图片描述

在技术栈分化的背景下，有两个趋势值得关注：

多模态能力成为关键：技术栈正从纯文本处理向融合图像、声音等多模态信息演进。
智能体（Agent）提升应用智能：基于LangChain等框架构建的智能体，能自主理解任务、调用工具并完成复杂目标。

人工智能的发展

虽然现在的AI已经出现了蓬勃发展的趋势，但是随着后面人工智能的不断发展，如何满足它天文数字一般的算力需求，是我们需要关注的。
以祖国来说，中国的日均Token消耗量从2024年初的1000亿，激增至2025年6月的30万亿，但是国产芯片在性能、功耗、生态上与国际领先水平存在差距。

三次浪潮

第一次浪潮（1950s-1970s）

这一浪潮始于阿兰·图灵提出的“机器能否思考”的哲学追问，并在1956年的达特茅斯会议上由约翰·麦卡锡等人正式确立“人工智能”学科。早期标志性成果是弗兰克·罗森布拉特提出的感知机，它被媒体誉为“会学习的机器”，引发了社会对“通用AI”即将到来的巨大热情。然而，马文·明斯基等学者很快证明了感知机的理论局限（如无法解决“异或”问题），加之当时计算机算力极其原始，无法支撑任何复杂任务。巨大的期望落空后，政府和资本迅速撤离，AI陷入了第一次寒冬。

第二次浪潮（1980s-1990s）

本次浪潮的核心是专家系统的商业化繁荣，例如爱德华·费根鲍姆团队开发的医疗诊断系统MYCIN。这类系统在特定领域表现出色，催生了第一批AI公司，并引发了如日本“第五代计算机计划”这样的国家级竞争。同时，杰弗里·辛顿等人成功应用的反向传播算法，让多层神经网络训练成为可能，为未来埋下了种子。但热潮最终因专家系统的“知识库”难以构建和维护（知识工程瓶颈）而消退，同时训练神经网络的算力需求远超当时硬件水平。随着个人电脑的兴起，昂贵的专用AI硬件被淘汰，AI再次陷入寒冬。

第三次浪潮（2000s-至今）

本轮浪潮的爆发源于三大要素的汇聚：互联网带来的大数据、GPU的广泛应用所提供的强大并行算力，以及深度学习算法的成熟。2012年，杰弗里·辛顿团队凭借AlexNet模型在ImageNet竞赛中取得压倒性胜利，震惊了整个学术界，标志着深度学习时代的到来。社会关注度空前高涨，从AlphaGo击败人类棋手到ChatGPT的横空出世，AI迅速成为全球科技竞争的核心焦点，并深刻改变着各行各业。然而，当前模型对算力的贪婪消耗也引发了人们对可持续发展、伦理和安全的新一轮担忧与反思。

从历史上来看，人工智能的发展也并不是一路高进的，他曾多次面临着迅速兴起，饱受社会的关注，却因为算力不足而陷入发展的停滞。

虽然现在的AI发展已成为时代的趋势所向，但是我们要学会尊重历史的法则，在学习技术和理论的同时，我们也应该去关注算力发展的领域。