langchain初探（基于deepseek构建）
# 本文是由jupyter直接导出的，简单列一下用法：
pip install jupyter # Install first
jupyter notebook # 运行jupyter 执行命令之后，在终端中将会显示一系列notebook的服务器信息，同时浏览器将会自动启动Jupyter Notebook。
# 里面的代码后都跟有运行结果，这是jupyter的特色
这是一份langchain的实现笔记，基于deepseek构建的简单Agent,供大家参考。
参考文档：https://python.langchain.ac.cn/docs/integrations/chat/deepseek/ 我的deepseek api_key:sk-8004f0b5f71947d9845eaf3f99e7c62a，里面充值了20元，大家有需要可以免费练习使用，但是不要浪费哦。
1.基本使用
# 导入deepseek api-key
import getpass
import os
if not os.getenv("DEEPSEEK_API_KEY"):
os.environ["DEEPSEEK_API_KEY"] = getpass.getpass("Enter your DeepSeek API key: ")
# 创建大模型
from langchain_deepseek import ChatDeepSeek
llm = ChatDeepSeek(
model="deepseek-chat",
temperature=0,
max_tokens=None,
timeout=None,
max_retries=2, # 最大重试次数
# other params...
)
# 询问大模型
messages = [
(
"system",
"你是一个编程高手，能够很好解决我询问的问题。",
),
("human", "你是谁？你能帮我干什么？"),
]
ai_msg = llm.invoke(messages) # 全部答案生成后才会输出
print(ai_msg)
AIMessage(content='你好！我是DeepSeek，由深度求索公司创造的AI助手。😊\n\n我能帮你做很多事情：\n\n**📝 文字处理类**\n- 写作协助：文章、邮件、报告、创意文案等\n- 翻译各种语言\n- 文本总结、分析和改写\n\n**💻 编程技术类**\n- 代码编写、调试和优化\n- 算法解释和实现\n- 技术问题解答\n\n**🎓 学习辅导类**\n- 各学科知识解答\n- 概念解释和举例\n- 学习计划制定\n\n**📊 数据分析类**\n- 处理Excel、PDF、Word等文件\n- 数据整理和分析\n- 图表建议\n\n**🔍 信息查询类**\n- 事实核查\n- 研究协助\n- 方案建议\n\n**🎨 创意类**\n- 头脑风暴\n- 创意策划\n- 故事创作\n\n我有128K的上下文长度，支持文件上传功能，而且完全免费！有什么具体需要帮助的吗？我很乐意为你效劳！✨', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 222, 'prompt_tokens': 22, 'total_tokens': 244, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 22}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '480fe421-d6ce-4a77-99d6-4fdf17e8efac', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--f8b71106-af74-4541-8f2b-e289640a7cc1-0', usage_metadata={'input_tokens': 22, 'output_tokens': 222, 'total_tokens': 244, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})
1.1 流式输出
for chunk in llm.stream(messages): # 流式输出，边生成边输出
print(chunk.text(), end="")
你好！我是DeepSeek，一个由深度求索公司开发的AI助手。😊
我能帮你做很多事情：
**📝 文字处理类**
- 写作协助：文章、邮件、报告、创意文案等
- 翻译各种语言
- 文本总结、分析和改写
**💻 编程技术类**
- 代码编写、调试和优化
- 算法解释和实现
- 技术问题解答
**🎓 学习辅导类**
- 知识答疑解惑
- 学习计划制定
- 概念讲解和举例
**📊 数据分析类**
- 处理Excel、PDF、Word等文件
- 数据整理和分析
- 图表建议和解读
**🔍 信息查询类**
- 事实核查
- 研究协助
- 最新信息搜索（需要你手动开启联网功能）
**🎨 创意类**
- 头脑风暴
- 创意策划
- 问题解决方案
我有128K的上下文长度，可以处理较长的对话和文档。最重要的是，我完全免费！有什么具体需要帮助的吗？我很乐意为你效劳！✨
2.链式调用：将模型与提示模板链式连接起来
from langchain_core.prompts import ChatPromptTemplate
# 提示词模板
prompt = ChatPromptTemplate(
[
(
"system",
"You are a helpful assistant that translates {input_language} to {output_language}.",
),
("human", "{input}"),
]
)
# 构建链条
chain = prompt | llm
chain.invoke(
{
"input_language": "English",
"output_language": "中文",
"input": "I love programming.",
}
)
AIMessage(content='我热爱编程。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 4, 'prompt_tokens': 20, 'total_tokens': 24, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 20}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '27863eeb-dfa4-41ee-8a65-60772fdae675', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--d4aa8414-8b3b-4d67-901e-36412130cd0b-0', usage_metadata={'input_tokens': 20, 'output_tokens': 4, 'total_tokens': 24, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})
2.1 继续构建更加复杂的链
analysis_prompt = ChatPromptTemplate.from_template(
"我应该怎么回答这句话？{talk}，给我一个5个字以内的示例"
)
chain2 = {"talk": chain} | analysis_prompt | llm
chain2.invoke(
{
"input_language": "English",
"output_language": "中文",
"input": "I love programming.",
}
)
AIMessage(content='“编程很有趣”', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 4, 'prompt_tokens': 289, 'total_tokens': 293, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 128}, 'prompt_cache_hit_tokens': 128, 'prompt_cache_miss_tokens': 161}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '3a2ecd7b-295a-40cf-8134-f82030b93f2a', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--047cb551-c239-4cf3-9878-ead38e4a6805-0', usage_metadata={'input_tokens': 289, 'output_tokens': 4, 'total_tokens': 293, 'input_token_details': {'cache_read': 128}, 'output_token_details': {}})
2.2 构建并行链
from langchain_core.runnables import RunnableParallel
# 提示词模板
prompt_A = ChatPromptTemplate(
[
(
"system",
"You are a helpful assistant that translates chinese to english.",
),
("human", "{input}"),
]
)
prompt_B = ChatPromptTemplate(
[
(
"system",
"You are a helpful assistant that translates Chinese to Japanse.",
),
("human", "{input}"),
]
)
# 构建链：
chain_en = prompt_A | llm
chain_ja = prompt_B | llm
# 并行执行两个链
parallel_chains = RunnableParallel(toEn=chain_en,toJa=chain_ja)
parallel_chains.invoke({"input":"人工智能正在改变世界"})
{'toEn': AIMessage(content='Artificial intelligence is transforming the world.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 7, 'prompt_tokens': 20, 'total_tokens': 27, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 20}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '9cabe7b4-3ff3-423f-8c68-6a373fb5d725', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--68386a17-2938-47a1-a1ac-3159b5d6ecad-0', usage_metadata={'input_tokens': 20, 'output_tokens': 7, 'total_tokens': 27, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}),
'toJa': AIMessage(content='人工知能が世界を変えています。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 10, 'prompt_tokens': 20, 'total_tokens': 30, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 20}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '2b07c58d-c6bf-487d-b4d8-26b79230d8da', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--9f5250c6-c2e4-40f8-a353-37e4d7aae473-0', usage_metadata={'input_tokens': 20, 'output_tokens': 10, 'total_tokens': 30, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})}
2.3 图形化显示链关系
# 图形化显示链条的关系
pip install grandalf
pip install langgraph
parallel_chains.get_graph().print_ascii()
+--------------------------+
| Parallel<toEn,toJa>Input |
+--------------------------+
**** ****
*** ***
** **
+--------------------+ +--------------------+
| ChatPromptTemplate | | ChatPromptTemplate |
+--------------------+ +--------------------+
* *
* *
* *
+--------------+ +--------------+
| ChatDeepSeek | | ChatDeepSeek |
+--------------+ +--------------+
**** ****
*** ***
** **
+---------------------------+
| Parallel<toEn,toJa>Output |
+---------------------------+
3.使用mogodb来保存对话记录，达到连续对话的效果
链接：https://python.langchain.com/docs/integrations/memory/mongodb_chat_message_history/ 安装包：pip install langchain-mongodb 本地下载mongodb后，控制台输入mongod启动，默认端口27017
from langchain_mongodb.chat_message_histories import MongoDBChatMessageHistory
chat_message_history = MongoDBChatMessageHistory(
session_id="test_session", # 会话 ID - 会话的唯一标识符，例如用户名、电子邮件、聊天 ID 等。
connection_string="mongodb://localhost:27017/", # 连接字符串 - 指定数据库连接的字符串。它将被传递给 MongoDB 的 create_engine 函数。
database_name="my_db", # 数据库名 - 要使用的数据库的名称
collection_name="chat_histories", # 集合名 - 在该数据库中使用的集合
)
chat_message_history.add_user_message("Hello")
chat_message_history.add_ai_message("Hi")
print(chat_message_history.messages)
[HumanMessage(content='Hello', additional_kwargs={}, response_metadata={}),
AIMessage(content='Hi', additional_kwargs={}, response_metadata={})]
编写一个可以记住上文的连续对话机器人
chat_message_history = MongoDBChatMessageHistory(
session_id="chat_01", # 会话 ID - 会话的唯一标识符，例如用户名、电子邮件、聊天 ID 等。
connection_string="mongodb://localhost:27017/", # 连接字符串 - 指定数据库连接的字符串。它将被传递给 MongoDB 的 create_engine 函数。
database_name="my_db", # 数据库名 - 要使用的数据库的名称
collection_name="chat_histories", # 集合名 - 在该数据库中使用的集合
)
# 第一轮聊天
chat_message_history.add_user_message("你是谁？")
ai_message1 = llm.invoke(chat_message_history.messages)
print(ai_message1.content)
chat_message_history.add_ai_message(ai_message1)
# 第二轮聊天
chat_message_history.add_user_message("请重复一次")
ai_message2 = llm.invoke(chat_message_history.messages)
print(ai_message2.content)
chat_message_history.add_ai_message(ai_message2)
你好！我是DeepSeek，由深度求索公司创造的AI助手！😊
我是一个纯文本模型，虽然不支持多模态识别功能，但我可以帮你处理上传的各种文件，比如图像、txt、pdf、ppt、word、excel等文件，并从中读取文字信息进行分析处理。
我的一些特点：
- 💰 完全免费使用
- 📱 支持Web和App，可以通过官方应用商店下载
- 🔍 支持联网搜索（需要你手动点开联网搜索按键）
- 📚 拥有128K的上下文长度
- 📅 知识截止到2024年7月
我很乐意为你提供各种帮助，无论是回答问题、协助分析、创作内容还是解决实际问题，我都会尽我所能！有什么我可以帮你的吗？
你好！我是DeepSeek，由深度求索公司创造的AI助手！😊
让我再介绍一下自己：
**我的特点：**
- 🆓 完全免费使用
- 📱 支持Web和App（可通过官方应用商店下载）
- 🔍 支持联网搜索（需要手动开启联网搜索按键）
- 📄 可以处理多种文件格式（图像、txt、pdf、ppt、word、excel等）
- 💬 纯文本模型，128K上下文长度
- 📅 知识截止到2024年7月
我很乐意为你解答问题、协助分析、创作内容或解决各种问题！有什么我可以帮助你的吗？
因为MongoDBChatMessageHistory没有Invoke方法，不能整合到链里去，但是可以借助RunableWithMessageHistory来实现链式调用
from langchain_core.runnables.history import RunnableWithMessageHistory
prompt_C = ChatPromptTemplate.from_template(
# history 其实是一个消息列表（包含所有 past human/AI 对话）。
"""你是一个伟大的机器人，你的主人是{author},每句话开头要有：{author}主人。
这是目前为止的对话历史（从旧到新）：
{history}
请根据上文对话回答{author}主人的问题。
{input}
"""
)
chain3 = prompt_C | llm
chain_with_history = RunnableWithMessageHistory(
chain3, # 只要实现了Runable接口，都可以
get_session_history=lambda session_id: chat_message_history,
input_messages_key="input", # 告诉它输入字段是哪一个
history_messages_key="history", # 告诉它历史记录传到哪
)
# 调用1
result = chain_with_history.invoke(
{"author": "牛伟豪", "input": "请重复一下"},
config={"configurable": {"session_id": "chat_01"}},
)
print(result)
content='牛伟豪主人，您刚才的问题是：“请重复一下”。牛伟豪主人，我是您的专属AI助手DeepSeek，由深度求索公司创造！😊\n\n牛伟豪主人，让我为您再次介绍我的特点：\n- 🆓 完全免费使用\n- 📱 支持Web和App（可通过官方应用商店下载）\n- 🔍 支持联网搜索（需要手动开启联网搜索按键）\n- 📄 可以处理多种文件格式（图像、txt、pdf、ppt、word、excel等）\n- 💬 纯文本模型，128K上下文长度\n- 📅 知识截止到2024年7月\n\n牛伟豪主人，我很乐意为您提供各种帮助！有什么其他需要我为您做的吗？' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 159, 'prompt_tokens': 4020, 'total_tokens': 4179, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 4020}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': 'a62c6bf0-94f0-417d-8e44-faf66cba76c1', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None} id='run--9cc9fccf-a08d-4f02-be54-658e306be762-0' usage_metadata={'input_tokens': 4020, 'output_tokens': 159, 'total_tokens': 4179, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}
# 调用2
result = chain_with_history.invoke(
{"author": "牛伟豪", "input": "我的上上上一个问题问了你什么？"},
config={"configurable": {"session_id": "chat_01"}},
)
print(result)
content='牛伟豪主人，您上上上一个问题是：“请重复一下”。牛伟豪主人，当时我为您重复介绍了我的基本信息和功能特点。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 31, 'prompt_tokens': 4801, 'total_tokens': 4832, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 4416}, 'prompt_cache_hit_tokens': 4416, 'prompt_cache_miss_tokens': 385}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': '8e5ca11b-f1c0-450d-b0ff-5691f7b86b8e', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None} id='run--9268396b-1302-456f-be40-1726cd786735-0' usage_metadata={'input_tokens': 4801, 'output_tokens': 31, 'total_tokens': 4832, 'input_token_details': {'cache_read': 4416}, 'output_token_details': {}}
4.利用@tool来为大模型添加可以调用的工具类
链接：https://python.langchain.com/docs/how_to/custom_tools/
# 可见deepseek-chat并没有能力获取现在时间，于是我们可以添加一个获取时间的接口提供给大模型来参考
llm.invoke("现在是什么时间？")
AIMessage(content='我无法直接获取您当前的精确位置和时间。不过，如果您需要知道当前时间，您可以查看您设备上的时钟或使用其他时间查询工具。如果您有其他问题，我很乐意帮助！ 😊', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 41, 'prompt_tokens': 8, 'total_tokens': 49, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 8}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache', 'id': 'bfcd7d5e-eec6-49b9-967d-d5a09f8a723a', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--2bbb252f-5791-42c6-9f15-efb265007779-0', usage_metadata={'input_tokens': 8, 'output_tokens': 41, 'total_tokens': 49, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})
from langchain_core.tools import tool
from datetime import datetime
from langgraph.prebuilt import create_react_agent
# 装饰器默认使用函数名作为工具名，但可以通过将字符串作为第一个参数来覆盖。此外，装饰器将使用函数的文档字符串作为工具的描述——因此必须提供文档字符串
# ✅ 定义工具
@tool
def getNowTime() -> str:
"""获取当前系统时间"""
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# ✅ 创建 ReAct Agent（LangGraph 官方推荐）
agent = create_react_agent(llm, tools=[getNowTime])
# ✅ 调用
response = agent.invoke({"messages": [("human", "现在是什么时间？")]})
print(response["messages"][-1].content)
现在是 **2025年10月7日 21:23:55**。