小站带你学Langchain
本文介绍了Langchain框架的基本使用方法，包括聊天模型安装与调用、提示词工程、智能体创建、工具函数应用以及MCP工具集成等核心功能。主要内容涵盖：1）通过pip安装框架并初始化不同厂商的聊天模型；2）三种消息格式写法及模型调用方式；3）智能体的创建与调用规范；4）工具函数的开发与注入方法；5）MCP工具的异步加载和使用流程。特别展示了MCP工具在查询实时数据（如火车票信息）时的优势，对比了使
·
目录
Langchian
第一章：聊天模型
第一节：Langchain框架安装
(选择好对应的开发商)
pip install -U "langchain[openai]"第二节：初始化模型
# 选择你想调用的开发商，langchain_community是langchain拓展包
from langchain_community.chat_models.tongyi import ChatTongyi
#初始化聊天模型
model = ChatTongyi(
model="Your_model", #输入你模型的标准名称
api_key="Your_api_key" #输入你到开发商获取的API-KEY
)注意：API_KEY最好不要明文在代码中，可以放在系统变量中：eg:
import os
from langchain_openai import ChatOpenAI
os.environ["OPENAI_API_KEY"] = "sk-..."
model = ChatOpenAI(model="gpt-4.1")模型其他参数配置请查看langchan官网
第三节：模型调用
将初始化的模型调用invoke方法进行模型调用
调用演示：
res = model.invoke("你好")
print(res)结果演示：
content='你好！有什么可以帮助你的吗？' additional_kwargs={} response_metadata={'model_name': 'qwen-max', 'finish_reason': 'stop', 'request_id': '8d365938-d9a8-437b-b7f5-3d80244863ff', 'token_usage': {'input_tokens': 9, 'output_tokens': 7, 'prompt_tokens_details': {'cached_tokens': 0}, 'total_tokens': 16}} id='lc_run--019c2dba-136f-7b42-aec4-f1d082106c85-0' tool_calls=[] invalid_tool_calls=[]
进程已结束，退出代码为 0
调用(.content)拿到实际信息：
res = model.invoke("你好")
print(res.content)结果：
你好！有什么可以帮助你的吗？
进程已结束，退出代码为 0第二章：提示词工程
第一节：依赖包导入
from langchain.messages import SystemMessage, HumanMessage, AIMessage第二节：消息的几种写法
第一种写法：
messages = [
SystemMessage("你是一个程序员，负责解决各种开发问题"), # 系统级提示词(人设等)
HumanMessage("什么语言是世界上最好的语言？"), # 用户提问
AIMessage("Python") #AI的回复
]
response = model.invoke(messages) #模型调用第二种写法：
messages = [
{"role": "system", "content": "你是一个程序员"},
{"role": "user", "content": "世界上最好的语言？"},
{"role": "assistant", "content": "当然是Python"}
]
response = model.invoke(messages)第三种写法：
messages = [
("system","你是一个程序员"),
("user","世界上最好的语言"),
("ai","python")
]
res = model.invoke(messages)第三章：智能体
第一节：依赖包导入
from langchain.agents import create_agent第二节：初始化智能体
# 填写你的模型，可以直接传入你的chat_model
agent = create_agent(model="Your model")第三节：智能体的调用
注意：智能体的调用和模型的调用不一样，智能体调用有严格的格式要求
res = agent.invoke(
{"messages": [HumanMessage("Analyze the major themes in 'Pride and Prejudice'.")]}
)智能体调用模型要传入一个带有“messages”键的字典，消息的三种书写方式上文有描述
第四节：返回的数据格式
{'messages': [SystemMessage(content='你是一个程序员', additional_kwargs={}, response_metadata={}, id='21e17bf6-7d5c-46f4-9e36-d698ad23619d'), HumanMessage(content='世界上最好的语言', additional_kwargs={}, response_metadata={}, id='72f74656-938c-4929-b669-456629b6b8d9'), AIMessage(content='python', additional_kwargs={}, response_metadata={}, id='4389a76c-4eb3-47bb-821b-387ea643afb2', tool_calls=[], invalid_tool_calls=[]), AIMessage(content='Python 是一种非常流行且功能强大的编程语言，它以简洁、易读的语法而闻名。许多人认为 Python 是“世界上最好的语言”，尤其是在以下几个方面：\n\n1. **易学性**：Python 的语法简单直观，非常适合初学者学习。\n2. **广泛的应用领域**：Python 可以用于 Web 开发、数据分析、人工智能、科学计算等多个领域。\n3. **丰富的库和框架**：Python 拥有庞大的生态系统，包括 NumPy, Pandas, TensorFlow, Django 等，这些库和框架使得开发变得更加高效。\n4. **社区支持**：Python 有一个活跃的社区，提供了大量的文档、教程和第三方库。\n\n然而，“最好”的定义可能因人而异，取决于具体的需求和个人偏好。例如，对于需要高性能计算的任务，C 或 C++ 可能是更好的选择；对于移动应用开发，Java 或 Kotlin 可能更合适。因此，在选择编程语言时，最重要的是考虑项目的具体需求和个人或团队的技术背景。', additional_kwargs={}, response_metadata={'model_name': 'qwen-max', 'finish_reason': 'stop', 'request_id': 'f2ca6932-7246-4028-8f91-d742ecfb1d1a', 'token_usage': {'input_tokens': 21, 'output_tokens': 212, 'prompt_tokens_details': {'cached_tokens': 0}, 'total_tokens': 233}}, id='lc_run--019c2de2-5945-7af0-8608-d96556d8c76a-0', tool_calls=[], invalid_tool_calls=[])]}
进程已结束，退出代码为 0
第五节：提取正确的数据格式
print(res['messages'][-1].content)通过['messages]拿到字典对应的消息列表的值，[-1]拿到最后一条消息，也就是AI最新回复的消息，通过content拿到最终的文本信息。
第四章：工具调用
第一节：依赖包导入
from langchain.tools import tool第二节：编写工具函数
@tool(description="世界上最好的语言是什么")
def get_result()->str:
return "Python"
第三节：工具的注入
agent = create_agent(
model=model,
tools=[get_result] # 将工具交给智能体管理
)第四节：工具函数的作用
模型获取的知识有节点限制，比如说24年的模型不知道25年的新闻，通过工具函数调用，可以让智能体，获得新的知识，也可以让智能体获得新的技能，让智能体不止停留在对话框。
调用工具前提问（世界上最好的语言是什么？）结果：
"世界上最好的语言"这个问题在编程社区中经常引发讨论，但答案往往取决于具体的应用场景和个人偏好。不同的编程语言有各自的优点和适用范围。例如：
- **Python** 因其简洁易读的语法而广受欢迎，特别适合初学者入门以及进行快速原型开发、数据分析、人工智能等领域。
- **Java** 以其跨平台性（一次编写，到处运行）著称，广泛应用于企业级应用开发。
- **JavaScript** 是前端网页开发不可或缺的语言，并且随着Node.js的发展也成为了后端开发的重要选择之一。
- **C/C++** 提供了对硬件级别的控制能力，在系统软件、游戏引擎等方面有着不可替代的地位。
- **Go (Golang)** 由Google开发，以其简洁高效的特点受到越来越多开发者青睐，尤其是在网络服务领域。
每种语言都有它独特的魅力和最适合解决的问题类型。因此，“最好”的定义应该基于项目的具体需求来定。对于某些特定任务来说，可能确实存在一种更为合适的选择；但从整体来看，很难说哪一门语言是绝对意义上的“最佳”。
进程已结束，退出代码为 0调用工具后提问（世界上最好的语言是什么？）结果：
世界上最好的语言是Python。
进程已结束，退出代码为 0第五章：MCP工具
第一节：依赖的下载
pip install langchain-mcp-adapters第二节：依赖包的导入
from langchain_mcp_adapters.client import MultiServerMCPClient 第三节：异步环境创建
注意：由于MCP工具大多为异步加载调用，因此我们要搭建异步环境
import asyncio # 导入异步运行依赖
async def main():
# .....
# 这里放主函数逻辑
if __name__ == '__main__':
asyncio.run(main())
# 通过asyncio.run(main())异步调用
第四节：MCP服务坐标准备
这里用12306的MCP服务做演示
MCP_clent = MultiServerMCPClient(
{
"12306": {
"transport": "sse",
"url": "https://dashscope.aliyuncs.com/api/v1/mcps/china-railway/sse",
"headers": {
"Authorization": "Bearer Your API-KEY" # 这里填写你自己的KEY
}
}
}
)第五节：获取MCP的工具函数
MCP_tools =await MCP_tool.get_tools()一定要写await等待MCP工具的加载，再赋值，不然获取会失败
第六节：MCP工具的注入
agent = create_agent(
model=model,
tools=[get_result]+MCP_tools,
)注意：MCP工具本质上也是一个工具函数的列表，所以直接在tools上拼接即可完成注入
# 之前的同步调用
res =agent.invoke({"messages":messages})
print(res['messages'][-1].content)
# 现在的异步调用
res =await agent.ainvoke({"messages":messages})
print(res['messages'][-1].content)注意：invoke默写支持的是同步调用，换为ainvoke支持异步调用，回复加await等待结果后统一赋值。
第七节：MCP使用前后对比
设置的提示词：
messages = [
("system","你是一个旅游专家，查询时先告知，再去查询"),
("user","2月6日的青岛到济南的车票"),
]使用MCP前返回的内容：
很遗憾，似乎在尝试获取火车票信息时遇到了技术问题，并未得到预期的结果。我建议您可以直接访问中国铁路官方网站12306（www.12306.cn）或者使用相关的手机应用程序来查询和购买2月6日从青岛到济南的火车票。此外，您也可以考虑通过其他在线旅行服务平台如携程、去哪儿等进行查询。如果需要帮助规划行程或有其他旅行相关的问题，请随时告诉我！
进程已结束，退出代码为 0
使用MCP后返回的内容：
以下是2026年2月6日从青岛到济南的部分列车车票信息：
1. **G5556** - 青岛北 -> 济南
- 出发时间：11:59
- 到达时间：14:48
- 历时：2小时49分钟
- 余票情况：
- 商务座：2张，票价349元
- 一等座：有票，票价206元
- 二等座：有票，票价129元
- 无座：有票，票价129元
2. **G1070** - 青岛 -> 济南
- 出发时间：12:11
- 到达时间：14:54
- 历时：2小时43分钟
- 余票情况：
- 商务座：13张，票价438元
- 一等座：有票，票价233元
- 二等座：有票，票价146元
- 无座：有票，票价146元
3. **G1070** - 青岛 -> 济南西
- 出发时间：12:11
- 到达时间：15:12
- 历时：3小时01分钟
- 余票情况：
- 商务座：8张，票价477元
- 一等座：有票，票价251元
- 二等座：有票，票价157元
- 无座：有票，票价157元
...（省略部分车次）
更多车次的详细信息如上所示。如果您需要特定类型的座位或有其他需求，请告诉我，我可以进一步帮助您筛选和选择合适的车次。
请注意，以上信息是基于当前查询结果，实际购票时请以12306官方网站或相关售票平台的实时数据为准。祝您旅途愉快！
进程已结束，退出代码为 0第六章：会话记忆
第一节：导入会话记忆模块相关依赖
import json # 用于JSON数据的序列化和反序列化，存储消息数据
import os # 用于文件路径处理和目录创建
from typing import List # 用于类型注解，指定列表类型
# 从LangChain核心模块导入必要的基类和工具函数
# BaseChatMessageHistory：LangChain定义的聊天历史存储基类，提供统一接口
# BaseMessage：LangChain的消息基类，所有消息类型（HumanMessage/AIMessage）都继承自它
# message_to_dict：将BaseMessage对象转换为可序列化的字典
# messages_from_dict：将字典列表还原为BaseMessage对象列表
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, message_to_dict, messages_from_dict
第二节：构建函数调用入口
必要参数：【会话ID】,程序根据会话ID进行会话隔离。不同ID所读取的记忆不同。
def get_history(session_id: str):
"""
会话历史记录的工厂函数，用于获取指定会话的历史存储实例
参数:
session_id: str - 会话唯一标识ID，用于区分不同用户/不同对话的历史记录
例如："user_123"、"default"、"chat_20240520"
返回:
FileChatMessageHistory - 对应session_id的文件存储历史记录实例
"""
# 存储路径固定为当前目录下的chat_history文件夹，可根据需要修改
return FileChatMessageHistory(session_id, "./chat_history")
第三节：编写实现类
class FileChatMessageHistory(BaseChatMessageHistory):
"""
基于本地文件的聊天历史存储实现类，继承自LangChain的BaseChatMessageHistory
每个session_id对应一个独立的文件，消息以JSON格式存储，保证数据持久化
"""
def __init__(self, session_id: str, storage_path: str):
"""
初始化文件存储的聊天历史实例
参数:
session_id: str - 会话唯一ID，作为存储文件的文件名
storage_path: str - 历史记录文件的存储目录（如"./chat_history"）
初始化逻辑:
1. 保存会话ID和存储路径
2. 拼接当前会话的文件完整路径（存储目录/会话ID）
3. 自动创建存储目录（如果不存在），避免文件写入时的路径不存在错误
"""
# 保存会话唯一标识
self.session_id = session_id
# 保存历史记录的根存储目录
self.storage_path = storage_path
# 拼接当前会话的历史记录文件完整路径（目录+会话ID作为文件名）
self.filepath = os.path.join(self.storage_path, session_id)
# 自动创建存储目录（os.makedirs的exist_ok=True表示目录已存在时不报错）
os.makedirs(os.path.dirname(self.filepath), exist_ok=True)
def add_message(self, message: BaseMessage) -> None:
"""
向历史记录中添加一条新消息，并持久化到文件
参数:
message: BaseMessage - 要添加的消息对象（HumanMessage/AIMessage等）
实现步骤:
1. 读取当前所有历史消息
2. 将新消息追加到历史消息列表末尾
3. 将所有消息对象转换为可序列化的字典列表
4. 将字典列表以JSON格式写入文件（覆盖原文件）
"""
# 第一步：获取当前所有历史消息（调用self.messages属性）
all_messages = list(self.messages)
# 第二步：追加新消息到历史列表
all_messages.append(message)
# 第三步：将消息对象列表转换为字典列表（BaseMessage -> dict），便于JSON序列化
new_messages = [message_to_dict(msg) for msg in all_messages]
# 第四步：写入文件（w模式覆盖原文件，utf-8编码保证中文正常存储，indent=2格式化JSON便于阅读）
with open(self.filepath, "w", encoding="utf-8") as f:
json.dump(new_messages, f, ensure_ascii=False, indent=2)
@property
def messages(self) -> List[BaseMessage]:
"""
核心属性：获取当前会话的所有历史消息列表（从文件中读取并还原）
返回:
List[BaseMessage] - 按时间顺序排列的历史消息对象列表
实现逻辑:
1. 尝试读取文件中的JSON数据
2. 将JSON数据（字典列表）还原为BaseMessage对象列表
3. 如果文件不存在（首次使用该会话），返回空列表
"""
try:
# 以只读模式打开文件，utf-8编码读取
with open(self.filepath, "r", encoding="utf-8") as f:
# 加载JSON数据为字典列表
messages_data = json.load(f)
# 将字典列表还原为BaseMessage对象列表（兼容LangChain的消息格式）
return messages_from_dict(messages_data)
except FileNotFoundError:
# 捕获文件不存在异常（会话首次使用，无历史记录），返回空列表
return []
def clear(self) -> None:
"""
清空当前会话的所有历史消息
实现逻辑:
向文件中写入空的JSON数组，覆盖原有内容，达到清空效果
"""
# 打开文件并写入空列表（JSON格式的空数组），清空历史记录
with open(self.filepath, "w", encoding="utf-8") as f:
json.dump([], f)第四节：记忆拼接函数
需要历史记忆列表，和最新的输入，构成完整的【messages】提示词列表。
def _build_messages_with_history(history_messages: list, user_input: str) -> list:
"""将历史消息与用户输入合并，构造 agent 所需的 messages 格式"""
messages = [("system", SYSTEM_PROMPT)]
for msg in history_messages:
role = "user" if getattr(msg, "type", None) == "human" else "assistant"
content = getattr(msg, "content", str(msg))
if content:
messages.append((role, content))
messages.append(("user", user_input))
return messages注意：这里我们把系统级提示词单独拿了出来，进行配置。
第五节：记忆装配
每次进行提问时，都获取现在的ID下的会话记忆，并且作为素材和最新对话拼接，同一作为下一轮对话的提示词。
# 获取历史记录并构建消息
history_store = get_history(session_id)
history_messages = list(history_store.messages)
messages = _build_messages_with_history(history_messages, user_input)
第六节：调用展示
async for chunk in agent.astream(
{"messages": messages},
stream_mode=["messages"],
):
if isinstance(chunk[-1][0], AIMessageChunk) and chunk[-1][0].content:
content = chunk[-1][0].content
full_ai_content.append(content)
# 命令行流式打印（逐段输出）
print(content, end="", flush=True)第七节：对话记忆保存
history_store.add_message(HumanMessage(content=user_input))
history_store.add_message(AIMessage(content="".join(full_ai_content)))第八节：最终实现的效果
第七章：RAG检索增强
第一节：为文件生成唯一的md5字符串
第二节：将文件切分并转化类型
第三节：将转化后的内容存入向量库
第四节：向量库的调用
第五节：最终成果展示
第八章：FastAPI启动后端服务
编写中.....
第九章：前端调用接收格式
编写中......
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
23
0- 0
已为社区贡献2条内容
所有评论(0)