在线API文档: https://api.python.langchain.com/en/stable/langchain_api_reference.html
0x00 TLDR
使用 deepseek、智谱 api 进行 langchain 开发学习。
0x01 基础依赖安装
# python -V
# Python 3.10.9
pip install langchian
pip install langchain_openai
pip install langchain_text_splitters
pip install zhipuai
pip install faiss-cpu
如遇到版本冲突，使用
pip install --upgrade xx把所有库都更新到最新版
0x02 代码示例
按顺序逐步引入高级应用
import os
import tabnanny
from typing import Any
# Deepseek 接口文档: https://api-docs.deepseek.com/zh-cn/
api_key = os.environ.get('DEEPSEEK_API_KEY')
base_url = 'https://api.deepseek.com'
model_name = 'deepseek-chat'
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
def demo_01_basic_usage():
"""
最简单的使用：通过OpenAI兼容协议，使用DeepSeek提供的会话接口，一次性返回结果
需要构建
"""
model = ChatOpenAI(api_key=api_key, base_url=base_url, model=model_name)
resp = model.invoke([
HumanMessage('简单回答: 为什么一周只有七天'),
]).content
print(resp)
from langchain.prompts import HumanMessagePromptTemplate, SystemMessagePromptTemplate
# 1.定义提示词模板
system_message_template = SystemMessagePromptTemplate.from_template(
"""你是一位专业的翻译官，可以把用户的输入翻译成{lang}语言。直接给出答案，不需要推理过程。"""
)
human_message_template = HumanMessagePromptTemplate.from_template(
"""{input}"""
)
def demo_02_work_with_template(lang, content):
"""
使用模版,可以定制提示词模型，提高使用灵活性
"""
# 2.构建消息, 对模板中的参数进行赋值
system_message = system_message_template.format(lang=lang)
user_message = human_message_template.format(input=content)
# 3.调用大模型
model = ChatOpenAI(api_key=api_key, base_url=base_url, model=model_name)
resp = model.invoke([
system_message,
user_message,
]).content
return resp
from langchain.prompts import ChatPromptTemplate
def demo_03_work_with_template_plus(lang, content):
"""
使用模版,可以定制提示词模型, 可以整合多个模板后，再一次性给入参数
"""
# 2.合并提示词模板
chat_prompt = ChatPromptTemplate.from_messages([system_message_template, human_message_template])
# 3.传入参数
prompt = chat_prompt.format_prompt(lang=lang, input=content)
# 4.调用大模型
model = ChatOpenAI(api_key=api_key, base_url=base_url, model=model_name)
resp = model.invoke(prompt).content
return resp
from pydantic import BaseModel, Field
from langchain.output_parsers import PydanticOutputParser
class UserInfo(BaseModel):
model_name: str = Field(default=['人名'], examples=['朱元璋'])
birthday: str = Field(default=['生日'], examples=['2000/02/28'])
def demo_04_json_output(username):
"""
格式化输出的结果为自定义对象json体
"""
# 1. 定义输出体
output_parser = PydanticOutputParser(pydantic_object=UserInfo)
# 2.定义提示词，需要在 system 提示词中使用 输出体
prompt = ChatPromptTemplate.from_messages(
[
('system', "{parser_instructions} 输出时使用中文。"),
('user', "请根据输入的历史人史，返回相应的名称、生日信息 {username}"),
]
)
final_prompt = prompt.invoke(
{"parser_instructions": output_parser.get_format_instructions(),
"username": username},
)
model = ChatOpenAI(api_key=api_key, base_url=base_url, model=model_name)
resp = model.invoke(final_prompt).content
return resp
from langchain.output_parsers import CommaSeparatedListOutputParser
def demo_05_chain_basic_usage(content):
"""
展示 langchain中的 chain 基础用法
"""
# 1.定义提示词模板
prompt = ChatPromptTemplate.from_messages([
('system', "{parser_instructions},仅输出结果"),
('human', "列出5个{subject}色系的十六进制颜色值"),
])
# 2.构建输出格式
output_parser = CommaSeparatedListOutputParser()
# 3.构建model
model = ChatOpenAI(api_key=api_key, base_url=base_url, model=model_name)
# 4.构建链
chain = prompt | model | output_parser
# 5.链式调用(invoke)，最后的参数，提供给最前面的 prompt
resp = chain.invoke({"subject": content, "parser_instructions": output_parser.get_format_instructions()})
return resp
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import ZhipuAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.schema import SystemMessage, HumanMessage
zhipu_key = os.environ.get('ZHIPU_API_KEY')
def demo_06_rag_basic(content=None):
"""
分几个步骤: 文档加载、分本分割、文本向量化、向量数据库存储、文档检索
"""
# 1.加载文本文件
loader = TextLoader("./demo.txt")
docs = loader.load()
# 2.分配分割
text_spliter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=40,
separators=["\n\n", "\n", "。", "！", "？", "，", "、", ""]
)
texts = text_spliter.split_documents(docs)
# 3.向量化模型：使用智谱AI
embedding = ZhipuAIEmbeddings(api_key=zhipu_key, model='embedding-3')
# 3.1 下面两行用于测试 向量化功能是否可用
# doc_results = embedding.embed_documents(["Hello", "world"])
# print(doc_results)
# 4.构建向量数据库. 采用FAISS内存数据库
db = FAISS.from_documents(texts, embedding)
# 5.检索
retriever = db.as_retriever()
retrieved_docs = retriever.invoke(content)
# print(retrieved_docs[0].page_content)
# 遍历前3个文档，并整合成参考文档上下文
db_content = "\n".join([docs.page_content for docs in retrieved_docs[:3]])
# 6.总结
model = ChatOpenAI(api_key=api_key, base_url=base_url, model=model_name, temperature=0)
messages = [
SystemMessage(content="""
请根据上下文回答问题，遵循规则：
1、如果上下文不相关，回答“未找到相关信息”
2、保持答案简洁，不超过3句话
"""),
HumanMessage(content=f"""上下文: {db_content}\n\n问题: {content}"""),
]
resp = model.invoke(messages).content
return resp
from langchain.agents import create_structured_chat_agent, AgentExecutor
from langchain.tools import BaseTool
from langchain import hub
class TextLengthTool(BaseTool):
name: str = "文本字数计算工具"
description: str = "当你被要求计算文本字数时，使用此工具"
def _run(self, text):
return len(text)
langsmith_key = os.environ.get('LANGSMITH_API_KEY')
def demo_07_agent_basic_text_len(content):
"""
向大模型提供自定义的工具
"""
# 1.定义 tool
tools = [TextLengthTool()]
# 2.获取官方提供的prompt, 这里需要:环境变量 LANGSMITH_API_KEY
prompt = hub.pull('hwchase17/structured-chat-agent', api_key=langsmith_key)
model = ChatOpenAI(api_key=api_key, base_url=base_url, model=model_name)
# 3.创建 agent
agent = create_structured_chat_agent(
llm=model,
tools=tools,
prompt=prompt,
)
# 4.创建agent执行体
agent_executor = AgentExecutor.from_agent_and_tools(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)
# 5.执行
resp = agent_executor.invoke({"input": content})
return resp
# from langchain_ollama import OllamaEmbeddings : 目前有Bug
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
ollama_base_url = os.environ.get('OLLAMA_API_URL') or 'http://localhost:11434/api/embeddings'
# pip install langchain-ollama
# pip install chromadb
def demo_08_ollama_embed_and_chroma_store(content):
"""
使用 ollama 提供的接口，进行embedding 实现
准备工作: ollama pull nomic-embed-text:v1.5
"""
# 1.创建ollama embedding对象
embedding = OllamaEmbeddings(base_url=ollama_base_url, model='nomic-embed-text:v1.5')
rst = embedding.embed_documents([content])
# 2.创建向量数据库
db = Chroma(collection_name='jira',
embedding_function=embedding,
persist_directory="./db")
# 3.加载本地文档
docs = TextLoader(file_path="./demo.txt").load()
# 4.向量化后存储
db.add_documents(docs)
# 5.顺带测试下
return db.similarity_search(content, k=2)
from langchain_community.llms import Ollama
def demo_09_ollama_model(content):
"""
使用Ollama 的deepseek进行单次请求
"""
# 1.构建模型
# model = Ollama(base_url=ollama_base_url, model='deepseek-r1:1.5b')
model = Ollama(base_url=ollama_base_url, model='deepseek-r1:7b')
# 2.测试
return model.invoke([SystemMessage(content='不需要推理过程，直接给出结果'), HumanMessage(content=content)])
import re
class RobotResp(BaseModel):
bug_type: str = Field(examples=["无法回充", "叠图"], description="扫地机故障分类")
def delete_think_tag(text: str) -> str:
rst = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL)
return rst.strip()
def demo_10_ollama_deepseek_json(content) -> RobotResp:
"""
使用Ollama deepseek,结构化返回结果，并处理一下去掉 <think></think>标签内容
"""
# 1.构建模型
model = Ollama(base_url=ollama_base_url, model='deepseek-r1:7b')
# 2.构建响应体分类
output_parser = PydanticOutputParser(pydantic_object=RobotResp)
# 3.构建提示词
prompt = ChatPromptTemplate.from_messages(
[
("system", "{parser_instructions}"),
("human", "你是一名专业的扫地机器人售后人员，擅长问题分类分析。请把用户问题进行分类：```{content}```")
]
)
final_prompt = prompt.invoke({"parser_instructions": output_parser.get_format_instructions(), "content": content})
# 4.调用大模型
resp = model.invoke(final_prompt)
# 5.使用对象进行解析response 并返回
try:
rr = output_parser.parse(resp)
return rr
except:
return resp
if __name__ == '__main__':
print("start run")
# demo_01_basic_usage()
# 一周七天源于古代文明对天文周期的观察：
#
# 1. ** 月相周期 **（约28天）被四等分，每段约7天。
# 2. ** 七大天体 **（太阳、月亮、金星、水星、火星、木星、土星）被古巴比伦人对应到每天，形成星期制。
# 3. ** 宗教文化 **（如《圣经》中上帝六日创世 + 第七日休息）强化了这一划分。
#
# 这一传统经罗马帝国、基督教传播被全球沿用，成为现代标准。
# print(demo_02_work_with_template('日语', '明天要加班么？'))
# 明日は残業ですか？
# print(demo_03_work_with_template_plus('日语', '明天不用加班'))
# 明日は残業なしです。
# print(demo_04_json_output('朱棣'))
# ```json
# {
# "model_name": "朱棣",
# "birthday": "1360/05/02"
# }
# ```
# print(demo_05_chain_basic_usage('姨妈红'))
# ['`#FF6B6B', '#FF8E8E', '#FFAAAA', '#FFC4C4', '#FFD8D8`']
# print(demo_06_rag_basic("卢浮宫有多少年历史了？"))
# 卢浮宫的建筑物始建于1190年左右，距今已有约830多年历史。
# print(demo_07_agent_basic_text_len("'把酒问青天',这句话几个字?"))
# > Entering new AgentExecutor chain...
# ```
# {
# "action": "文本字数计算工具",
# "action_input": {
# "text": "把酒问青天"
# }
# }
# ```5```
# {
# "action": "Final Answer",
# "action_input": "'把酒问青天'这句话共有5个字。"
# }
# ```
#
# > Finished chain.
# {'input': "'把酒问青天',这句话几个字?", 'output': "'把酒问青天'这句话共有5个字。"}
# print(demo_08_ollama_embed_and_chroma_store(content="卢浮宫有多少年历史"))
# [Document(metadata={'source': './demo.txt'}, page_content='罗浮宫（法语：Musée du Louvre,英语 /ˈluːv(rə)/ ），正式名称为罗浮博物馆，位于法国巴黎市中心的塞纳河边，原是建于12世纪末至13世纪初的王宫，现在是一所综合博物馆，亦是世界上最大的艺术博物馆之一，以及参观人数最多的博物馆，是巴黎中心最知名的地标。\n\n罗浮宫的建筑物始建于1190年左右，并在近代曾多次进行扩建，今天所见的模样则一个巨大的翼楼和亭阁建筑群，主要组成部分的总面积则超过60,600平方公尺（652,000平方英尺），馆内永久收藏则包括雕塑、绘画、美术工艺及古代东方、古代埃及和古希腊罗马等7个分类，主要收藏1860年以前的艺术作品与考古文物，罗浮宫博物馆在1793年8月10日开幕起正式对公众开放，平均每天有15,000名游客到此参观，其中65%是外国游客。\n\n位置\n\n罗浮宫与杜乐丽花园的卫星照片\n罗浮宫博物馆位于巴黎市中心的卢浮宫内，位于塞纳河右岸，毗邻杜乐丽花园。最近的两个地铁站是皇家宫-罗浮宫站和卢浮-里沃利站，前者有直达地下购物中心 Carrousel du Louvre 的地下通道。\n\n在1980年代末和1990年代大改建之前，罗浮宫共有好几个街道入口，目前大部分入口已经永久关闭。自1993年以来，博物馆的正门位置位于拿破仑广场金字塔底下的地下空间，游客可以从金字塔本身、旋转阶梯处连接到博物馆的通道。\n\n博物馆的参观时间随著时代的推移而变化。自18世纪开放以来，只有艺术家和来自外国的观光游客享有特权参观，这项特权后在1850年代才消失。当博物馆从1793年首次开放时，新历法法国共和历规定了“十天周”（法语：décades），其中前六天为艺术家和外国人访问，后三天为将军访问，民众仅能在最后一天参观，后在在1800年代初期在恢复七天周后，民众在每周只有4小时的时间能在罗浮宫参观，周六和周日则是缩减至下午2点至下午4点期间参观。\n\n从1824年开始的一项新规定允许公众在星期日和节假日时参观，然而其他日子只对艺术家和外国游客开放，这种情况到1855年才发生了变化，博物馆更改成除了周一外全天向公众免费开放，直到1922年才开始收费。\n\n当前自1946年开始，罗浮宫除了在周二公休和特殊假日外，通常向游客全面开放参观，内部允许使用照相机和录像机，但禁止使用闪光灯。')]
# print(demo_09_ollama_model('天为什么是蓝色。 请直接回答，不要包含 <think> 标签'))
# rst = demo_10_ollama_deepseek_json("扫地机无法进行回充")
# if type(rst) is RobotResp:
# print(rst.bug_type)
# else:
# print(f"parse failed: {rst}")
