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），让组件可以像管道一样连接
• 序列化协议：统一配置和状态的序列化标准
• 事件系统：提供组件间的通信机制

架构设计原理：基于接口的组件化系统

LangChain实现组件无缝集成的核心机制基于面向接口编程和适配器模式：

1. 标准化接口定义

   • 通过抽象基类为各类组件定义统一的调用契约
   • BaseLLM 规定语言模型必须实现 _call() 和 _generate() 方法
   • BaseTool 规定工具组件必须实现 _run() 方法
   • BaseRetriever 规定检索器必须实现 _get_relevant_documents() 方法

2. 接口与实现分离

   • 上层应用（Chains、Agents）仅依赖抽象接口，不依赖具体实现
   • 框架核心通过依赖注入机制动态绑定具体组件
   • 组件替换对业务逻辑透明，支持运行时热插拔

3. 适配器封装模式

   • 新模型集成时，开发者实现对应的接口包装类
   • 在包装类中封装原生API调用，适配标准接口规范
   • 输出格式统一转换为LangChain内部数据结构

LangChain 规定，不管什么LLM模型或工具，都必须提供一个标准插头，即实现统一的调用方法，比如 .call()。而LangChain框架的核心层就是一个只认标准的抽象接口的插座，如果有新的模型，LangChain就会让你为它配一个包装类，将模型原本不规则的接口转化为标准的接口，也就是说，LangChain是通过让大家遵守同一套调用规则（通过通过抽象基类定义的标准接口），并要求所有组件都通过一个适配器（包装类）来遵守这套规则，实现了不同组件之间的即插即用

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个）

此图片来源：机器学习社区

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模型,方便大家的练习。

推荐使用智谱AI的GLM-4-Flash模型（免费且响应快）：

1. 访问智谱AI开放平台
2. 注册账号并完成实名认证
3. 进入"控制台" → “API密钥”
4. 点击"创建新的API密钥"
5. 复制生成的API Key

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

1. 访问腾讯云TI平台
2. 注册腾讯云账号
3. 进入控制台申请API访问权限

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发展已成为时代的趋势所向，但是我们要学会尊重历史的法则，在学习技术和理论的同时，我们也应该去关注算力发展的领域。

• 量子计算
• 互联技术
• 软件与生态
• 芯片革新

最后

在这个快速演进的时代，保持技术敏感度固然重要，但同样重要的是保持人文关怀和批判性思维。技术的最终价值不在于其复杂程度，而在于它如何增强人类的能力、丰富人类的生活、促进人类的福祉。我们应该时刻保持更全面的思考。

非常感谢你的阅读，不介意的话，想麻烦你点赞或者关注，这会成为我不断更新的动力，有任何的讨论也可以在评论区提出，祝君顺遂。