Python后端开发之旅(四)

补充知识

Notebook笔记本(Jupyter/Colab)

在线 Notebook

登录之后:
https://anaconda.com/app/

安装 Jupyter

  • Vscodejupyter 插件交互没有 浏览器的好,而且也不能安装额外 的插件,更不能进行 演示 slide 的功能
    • 只能临时作为 不得已的交互方式
  • obsidian 的 slide演示功能 画风比较黑暗,而且内容过多的话 容易溢出屏幕,可以通过修改 css 进行,但是学习成本有点高
  • 所以着手 安装 jupyter
UV通过安装依赖

UV 下载以及相关操作

# vscode 的终端bash中操作
## 进入虚拟环境
uv venv --python /e/Lang/Python/Cpython/cpython-3.11.14-windows-x86_64-none
source .venv/Scripts/activate 

## 开始安装依赖包
uv init
uv add jupyter --default-index https://pypi.tuna.tsinghua.edu.cn/simple
或者想要快速安装
uv pip install jupyter --default-index https://pypi.tuna.tsinghua.edu.cn/simple 
或者没有uv的话
pip install jupyter -i https://pypi.tuna.tsinghua.edu.cn/simple

## 启动后端服务
jupyter --version
jupyter notebook
jupyter notebook /e/Lang  # 修改启动目录
jupyter notebook /d/Microsoft/个人简历/教书育人 # 授课路径

# 直接终端中——可能需要NPM(Node Package Manager)?
## 文件夹管理器
win + e -> E:\Lang\LLM\.venv\Scripts 
bash
jupyter notebook
## cmd
cd /d E:\Lang\LLM\.venv\Scripts
jupyter notebook
### 或者直接用参数形式启动
jupyter notebook D:\Microsoft\个人简历\教书育人
jupyter notebook . # 当前目录启动
通过 Anaconda——View/Cell Toolbar/Slideshow

一般使用anaconda自带的Anaconda Promp进行操作

Anaconda 安装以及 改变 Jupyter 的启动目录

启动Slide模式:点击这里

“魔法命令”(Magic Commands)

  • 是一些以百分号(%)或惊叹号(!)开头的特殊命令,用于执行一些与代码运行环境相关的操作
  • % 开头的魔法命令分为两类:行魔法命令(Line Magic)和单元魔法命令(Cell Magic)。行魔法命令以单 % 开头,作用于单行代码;单元魔法命令以双 %% 开头,作用于整个代码单元
  • ! 开头的命令用于在 Jupyter Notebook 中执行系统命令,类似于在终端中运行命令
  • %lsmagic 列出所有可用的魔法命令
  • !调用shell(在新进程中),而%影响与笔记本电脑相关的进程(或笔记本电脑本身;许多%命令没有shell对应项)
    • !cd foo本身没有持久效果,因为更改目录的进程会立即终止。
    • %cd foo更改笔记本电脑进程的当前目录,这是一种持久效果
  • 对于pip命令的区别:
    • % 运行shell的环境是当前jupyter运行的虚拟环境比如kernel 是pytorch,输入%pip list,就会显示当前虚拟环境安装的库
    • ! 运行shell的环境是jupyter的base主环境,输入!pip list,就会显示主环境安装的库

Python库和方法

常见的语法糖

PyMuPDF、pypdfium2、pdfplumber、pdfminer

  • pdfminer最早的Python PDF处理库之一
  • pdfplumber专注于PDF文本和表格提取,建立在pdfminer.six的基础上,提供更完整的表格提取功能
  • pypdfium2是PDFium库的Python绑定,PDFium是Google Chrome浏览器使用的PDF引擎
  • PyMuPDF是MuPDF库的Python绑定,基于C++的MuPDF库,性能极高
PDF处理生态系统
├── 底层引擎
│   ├── MuPDF (C++) → PyMuPDF
│   ├── PDFium (C++) → pypdfium2
│   └── pdfminer (Python) → pdfplumber
├── 功能定位
│   ├── 全功能处理:PyMuPDF, pypdfium2
│   ├── 文本提取:pdfplumber, pdfminer
│   └── 表格提取:pdfplumber
└── 性能层次
    ├── 高性能:PyMuPDF, pypdfium2
    └── 中等性能:pdfplumber, pdfminer
  • 使用 PyMuPDF(fitz)可以高效地从PDF中提取嵌入图片,还可以将整页渲染为图片,通常还会搭配使用Pillow用于保存图片

json

Json库供用户处理Json格式数据,包括4种方法:

  • dumps()——将python对象转成json格式的字符串
  • loads()——对字符串进行操作的
  • dump()——将python对象转成json格式存入文件
  • load()——对文件进行操作
    • 为了方便记忆,可以把loads后面的小尾巴s理解为str
  • 不是文件名后缀为.json的才属于json文件,无论有没有后缀,或者后缀是.txt等,只要文件内容符合下面的格式,都可以使用这些函数
    • dict:{“姓名”: “张三”, “年龄”: 18}
    • list:[“张三”, “李四”]
    • 字符串:“张三李四王五” (注意,必须有双引号)
    • 纯数字:123
  • json.dumps()json.dump()的详细参数
    • ensure_ascii 参数
      • 使用ensure_ascii=False输出原始Unicode字符
      • 默认情况下,ensure_ascii=True,非ASCII字符会被转义
    • indent 参数
      • 使用indent参数进行美化输出
      • 默认为 None,用于指定每个层级的缩进空格数,此时输出的JSON数据将尽可能紧凑
  • json.loads()json.load()的详细参数

tempfile

  • 标准库中的一个模块,用于创建临时文件和目录
    • 适用于需要临时存储数据的场景。临时文件和目录在使用后可以自动清理,避免手动管理的麻烦
  • 自动清理: TemporaryFileNamedTemporaryFileTemporaryDirectorySpooledTemporaryFile 是带有自动清理功能的高级接口,可用作上下文管理器
  • 手动清理: 使用 mkstempmkdtemp 创建的文件或目录需要手动删除
  • 比如shutil.rmtree(temp_dir)可以用于清理 文件夹

shutil(shell utility)

  • 标准库中的一个高级文件操作模块,主要用于文件和目录的复制、移动、删除、压缩和解压等操作。它是对os模块的补充,提供了更高层次的文件操作功能
  • 在这里插入图片描述
  • 保持了 Python “batteries included” 的理念

python-dotenv

  • python-dotenv 是一个用于从 .env 文件读取键值对并加载为环境变量的库,常用于管理敏感信息和配置,符合 12-factor 应用原则
  • 可以把所有用到的环境变量写到.env文件里,然后以k,v的方式读取作为环境变量
  • 当前目录或其父目录中的.env文件指定的路径加载环境变量
pip install  python-dotenv
# 自动搜索 .env 文件
load_dotenv()

# 或指定路径
# load_dotenv(dotenv_path="config/.env", override=True)

# 获取变量
db_url = os.getenv("DATABASE_URL")
secret = os.getenv("SECRET_KEY")
print(db_url, secret)

# 对于多环境配置,可按优先级加载
for env_file in (".env", ".env.local"):
   load_dotenv(env_file)

re

  • Python 的 re 模块是用于处理正则表达式的标准库模块。
  • 正则表达式(Regular Expression,简称 regex 或 regexp)是一种强大的工具,用于匹配、搜索和操作文本
  • re.match() 函数用于从字符串的起始位置匹配正则表达式。如果匹配成功,返回一个匹配对象;否则返回 None
  • re.search() 函数用于在字符串中搜索正则表达式的第一个匹配项。与 re.match() 不同,re.search() 不要求匹配从字符串的起始位置开始
  • re.findall() 函数用于查找字符串中所有与正则表达式匹配的子串,并返回一个列表
  • re.sub() 函数用于替换字符串中与正则表达式匹配的部分
  • match.group()Match对象 提取匹配结果的核心方法。当使用 re.match()re.search() 成功匹配后
    • group()group(0):返回完整匹配结果
    • group(n):返回第 n 个捕获组的内容(从 1 开始编号)
    • group(name):返回命名捕获组的内容
    • groups()方法返回一个元组,其中包含所有分组的匹配结果
# 将多个空白字符(包括换行符和制表符)替换为单个空格
text = re.sub(r'\s+', ' ', text)

# 修复常见的OCR问题,将制表符和换行符替换为空格
text = text.replace('\\t', ' ')
text = text.replace('\\n', ' ')
# OCR 常把换行、段落、表格中的间隔当成不同的控制字符,这会造成后续 chunk_text 切片、BM25 分词等逻辑看到不连贯或多余的空白
# 再用 text.split() 去重空格,可以恢复句子连续性

# 去除首尾空白字符,并确保单词之间只有一个空格
# 让单词之间只有一个空格,有助于后续分块和BM25分词的准确性
text = ' '.join(text.split())

nest-asyncio

  • nest_asyncio 是一个轻量级 Python 库,用于修补解决 Python asyncio 事件循环在嵌套场景下的限制。它允许在已经运行的事件循环中再次运行事件循环
    • 允许嵌套运行异步代码
    • Jupyter Notebook 中运行异步代码
  • 特别适合在 Jupyter Notebook、交互式 shell 或需要嵌套异步操作的场景中
    • 嵌套事件循环:允许在现有 asyncio 事件循环中运行新的异步代码,解决 RuntimeError: This event loop is already running
pip install nest-asyncio
# 修补 asyncio
nest_asyncio.apply()

async def inner_task():
    await asyncio.sleep(1)
    print("Inner task completed")

async def outer_task():
    print("Outer task started")
    await inner_task()
    print("Outer task completed")

# 在已有事件循环中运行
loop = asyncio.get_event_loop()
loop.run_until_complete(outer_task())

np.linalg.norm()

  1. linalg=linear(线性)+algebra(代数),norm则表示范数。
  2. 函数参数
x_norm=np.linalg.norm(x, ord=None, axis=None, keepdims=False)
  • ①x: 表示矩阵(也可以是一维)
  • ②ord :范数类型
  • ③axis:处理类型
    • axis=1表示按行向量处理,求多个行向量的范数
    • axis=0表示按列向量处理,求多个列向量的范数
    • axis=None表示矩阵范数。
  • ④keepding:是否保持矩阵的二维特性
  • 默认参数ord=None,axis=None,keepdims=False’
    在这里插入图片描述
def cosine_similarity(vec1, vec2):
    """
    计算两个向量之间的余弦相似度。
    参数:
    vec1 (np.ndarray): 第一个向量。
    vec2 (np.ndarray): 第二个向量。
    返回:
    float: 两个向量之间的余弦相似度。
    """
    # 计算两个向量的点积,并除以它们范数的乘积
    return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

Python进阶——LangChain

在这里插入图片描述

✅ 1. LangChain 0.1.x vs 1.0:版本演进与主要区别

LangChain 的第一个稳定版本,即 LangChain 0.1.0,于 2024 年 1 月 8 日正式发布,这是一个值得庆祝的里程碑,也是 LangChain 项目的一个新的起点
LangChain 的版本号由三个部分组成,即主版本号、次版本号和修订号,分别表示 LangChain 的大的、中等的和小的更新。例如,LangChain 0.1.0 表示 LangChain 的第 0 个主版本,第 1 个次版本,第 0 个修订版本

特性 LangChain 0.1.x(旧版) LangChain 1.0(新版)
发布时间 2024 年 1 月 8 日 2025年10月23日
核心理念 模块松散,流程不统一 统一、可组合、强类型

🔍 关键总结:

  • 0.1.x → 多个独立类(如 LLMChain, AgentExecutor
  • 1.0 → 统一 Runnable 接口,所有组件都是 Runnable模块重组,代码也得改变

✅ 2. 快速入门 LangChain(1.0)

📦 1. 安装

pip install langchain langchain-openai  # 以 OpenAI 为例

💡 你也可以用其他 LLM 提供商,如 Anthropic、Cohere、Claude、Google Vertex AI ,DeepSeek等


🧩 2. 核心概念图谱(类比 Java)

LangChain 概念 类比 Java 开发 说明
LLM RestTemplate / FeignClient 调用大模型 API
PromptTemplate String.format() + @Value 动态生成提示词
Runnable Function<T, R> / Supplier<T> 所有组件都实现此接口
Chain Service + Controller 逻辑处理 链
Tool @Component + @Autowired 工具函数,用于 Agent
Memory Session / Cache 记忆历史对话
LCEL Spring Expression Language 链式表达式语法 (`
Callback Interceptor / AOP 日志、监控、追踪

chain = {“context”: 本地Rag或者工具搜索 , “input”: RunnablePassthrough()} |prompt | llm | JsonOutputParser()

  • Conversational(Agent):用于与用户进行自然对话,并接收用户提出的问题,根据提问内容选择具体使用的处理工具
  • RetrievalQA(Retriever):用于从自定义资料集中检索相关内容。
  • DuckDuckGoSearchRun(tool): 用于访问DuckDuckGoSearch搜索API
  • Prompts & Structured output parser(ModelIO):用于定义接收对话的模板以及结构化搜索引擎查询到的结果并生成一个简洁和准确的回答。
  • SequentialChain(Chain):用于按照顺序执行以上组件,并将每个组件的输出作为下一个组件的输入
  • ConversationBufferMemory(Memory):用于在对话中记住以前的交互置,以便在后续的对话中使用。
  • StdOutCallbackHandler: 标准输出用于进行日志记录。

🧱 3. 基础组件详解

  • 大部分都是通过 lanchain_core 进行导入的
  • 通过from langchain_community.tools import DuckDuckGoSearchRun 会导入 社区的工具包
  • 还有一部分通过 lanchain 导入的是 比较顶层的 工具了,比如from langchain.callbacks import StdOutCallbackHandler 进行调式跟踪
✅ 3.1 LLM(大语言模型)
使用 OpenAI(最常见)
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.7)

📝 你可以通过 os.getenv("OPENAI_API_KEY") 读取密钥

自定义 LLM 类(高级用法)——invoke()
from langchain_core.runnables import Runnable

class CustomLLM(Runnable):
    def invoke(self, input: str, config: dict = None):
        # 实现自己的大模型调用逻辑
        return f"Custom response to: {input}"

# 使用
custom_llm = CustomLLM()
result = custom_llm.invoke("Hello")
print(result)

✅ 这样就可以把你的私有模型或本地模型接入 LangChain!


✅ 3.2 提示模板(PromptTemplate)

使用format()进行填入

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template(
    "You are a helpful assistant. Answer the question: {question}"
)

# 生成最终提示
formatted_prompt = prompt.format(question="What is Python?")
print(formatted_prompt)
# 输出: You are a helpful assistant. Answer the question: What is Python?

✅ 支持变量替换,可用于动态生成提示


✅ 3.3 工具(Tools)和工具包(Toolkits)
自定义工具(类似 Java 的 Service)
from langchain_core.tools import tool
import requests

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    url = f"https://api.weather.com/forecast?city={city}"
    response = requests.get(url)
    return response.json()["temperature"] if response.ok else "Error"
工具包(Toolkit)
from langchain_community.tools import DuckDuckGoSearchRun

search_tool = DuckDuckGoSearchRun()

✅ 工具可以被 Agent 调用,实现“查询互联网”功能


✅ 3.4 消息(Messages)与历史记录(Memory)
消息格式(类似聊天上下文)
  • 消息是LangChain中对话交互的基本单元,分为两类:

    • HumanMessage:表示用户输入(如提问或指令)
    • AIMessage:表示AI模型的响应
from langchain_core.messages import HumanMessage, AIMessage

messages = [
    HumanMessage(content="Hello"),
    AIMessage(content="Hi! How can I help?")
]
记忆机制(Memory)
  • 记忆模块用于持久化存储和管理对话历史,常见类型包括:

    • ConversationBufferMemory:完整保存所有对话记录
    • ConversationBufferWindowMemory:仅保留最近N轮对话
    • ConversationSummaryMemory:用LLM摘要压缩长对话
from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory()
memory.chat_memory.add_user_message("Hello")
memory.chat_memory.add_ai_message("Hi! How can I help?")
  • 保存对话历史供后续调用(如多轮对话上下文)
  • 支持通过memory.load_memory_variables()读取历史

✅ 类似 Session,但持久化更灵活


✅ 3.5 输出解析(Output Parsing)
目标:让模型返回结构化数据(如 JSON 使用 JsonOutputParser())
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate

parser = JsonOutputParser()

prompt = PromptTemplate(
    template="Answer only in JSON format: {question}",
    input_variables=["question"]
)

chain = prompt | llm | parser
result = chain.invoke({"question": "Who is the CEO of Tesla?"})
print(result)
# 输出: {"name": "Elon Musk"}

✅ 避免模型返回自由文本,提升结构化能力


✅ 3.6 回调处理(Callbacks)
用于日志、追踪、调试
from langchain.callbacks import StdOutCallbackHandler

callbacks = [StdOutCallbackHandler()]

llm.invoke("Hello", callbacks=callbacks)

✅ 可扩展为:LangChainTracerLangSmithCallbackHandler(企业级追踪)

LangChain 借助 LangSmith 提供了更好的日志、可视化、播放和跟踪功能,以便监控和调试 LLM 应用。LangSmith 是一个基于 Web 的工具,用于可视化和控制 LangChain 的链和代理,能够查看和分析 细化到class的输入和输出


✅ 3.7 LCEL(LangChain Expression Language)

| 链接模块,像流水线一样执行
就像 unix 的管道机制

示例:一个简单的 LCEL 流程
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda, RunnableSequence

# 1. 提示模板
prompt = PromptTemplate.from_template("Translate this to English: {text}")

# 2. Lambda 函数:转成大写
uppercase = RunnableLambda(lambda x: {"text": x["text"].upper()})

# 3. 链接:先转大写,再翻译
chain = uppercase | prompt | llm

# 执行
result = chain.invoke({"text": "你好世界"})
print(result.content)
  • RunnableLambda:将普通 Python 函数包装为 LangChain 可执行的组件
  • RunnableBranch 是 LangChain 中用于根据条件选择不同执行路径的组件
    • RunnableBranch 接受多个条件判断对(condition, component)以及一个默认的 fallback 函数
    • 每个条件判断对的结构是:(condition, component),其中:
      • condition 是一个函数,用于判断输入是否满足某个条件。
      • component 是一个可执行的组件(如 RunnableLambda),如果条件满足,则会执行该组件。
      • 如果所有条件都不满足,则执行最后一个 fallback 函数
更复杂的例子:带条件判断
from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: x["text"].startswith("hello"), lambda x: "Hello!"),
    (lambda x: x["text"].startswith("hi"), lambda x: "Hi!"),
    lambda x: "Unknown greeting"
)

result = branch.invoke({"text": "hello world"})

✅ 3. 调用方式区别

  • invoke():一次性同步调用,等待完整输出后返回。
  • stream():流式异步调用,边生成边返回。就是流式响应!!!
    • LangChain会将这些结果作为生成器逐步返回,开发者可在接收到每个分片(chunk)时进行实时处理或显示
      在这里插入图片描述
  • 有时我们既想实时输出,又想保存完整结果:
chunks = []
for chunk in llm.stream("请解释RAG的原理。"):
    print(chunk.content, end="", flush=True)
    chunks.append(chunk.content)

full_output = "".join(chunks)

✅ 4. 搜索引擎提供上下文:构建一个智能问答助手

🎯 功能:

  • 接收用户问题
  • 使用 LLM 生成回答
  • 支持简单工具(如搜索)
  • 保留聊天历史

🛠️ 代码结构

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableSequence
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_community.utilities import WikipediaAPIWrapper
from langchain_core.output_parsers import StrOutputParser

# 1. 初始化 LLM
llm = ChatOpenAI(model="gpt-4-turbo")

# 2. 定义工具
search_tool = DuckDuckGoSearchRun()
wikipedia = WikipediaAPIWrapper()

# 3. 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant.请根据以下上下文回答问题:{context}"),
    ("human", "{question}")
])

# 4. 构建 LCEL 流程
chain = (
    {
        "question": RunnablePassthrough(),
        "context": RunnableLambda(lambda x: search_tool.invoke(x["question"]))
    }
    | prompt
    | llm
    | StrOutputParser()
)

# 5. 运行
result = chain.invoke({"question": "Who is Elon Musk?"})
print(result)
  • 使用 LangChain 表达式语言(LCEL)构建处理流程:
    1. 接收输入问题并原样传递(RunnablePassthrough
    2. 使用搜索工具获取问题相关上下文
    3. 将问题和上下文传递给提示模板
    4. 模板输出传递给LLM
    5. 最后使用字符串输出解析器
  • 使用LangChain构建一个结合外部搜索的问答系统。
  • 检索到的信息会作为上下文提供给LLM,帮助生成更准确的回答

与直接调用OpenAI API 区别

维度 LangChain集成 直接OpenAI API调用
架构设计 模块化框架,支持链式操作和Agent决策 单点API请求,需手动处理上下文
数据处理 内置文档加载/向量检索/记忆管理 自行实现数据预处理和存储逻辑
开发效率 通过预置组件快速搭建复杂流程(如RAG) 适合简单问答场景,复杂逻辑需重复造轮子
多模型支持 可混合调用不同模型(如DeepSeek+Stable Diffusion) 单一模型交互,跨模型协同需额外开发
适用场景 知识库问答/自动化办公/复杂决策系统 简单文本生成/翻译/基础对话
可维护性 标准化组件,便于团队协作和代码维护 需要自定义实现,可维护性依赖于代码质量
扩展性 易于添加新功能和集成新模型 扩展需要修改核心代码

在需要精细控制部分环节时,可组合使用两者(如用LangChain处理数据流,关键节点直接调API)

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐