PythonOllamaLinux网络编程

大模型开发基础学习笔记

这份笔记整理了大模型应用开发需要用到的一组前置知识，包括本地模型、Python 基础、流程控制、常用容器、函数、面向对象、网络与并发、异常与文件处理和 Linux 命令。

一、本地智能聊天机器人

1. 整体结构

一个本地智能聊天机器人可以拆成前端和后端两部分：

• 前端负责接收用户输入并展示回答，例如 Chatbox 或 Streamlit。
• 后端负责运行模型并提供 API，例如 Ollama。
• 前端通过 HTTP 请求调用 Ollama 提供的服务。

2. HTTP 通信过程

用户在前端点击按钮后，前端会向后端发送 HTTP 请求：

一个 HTTP 请求由请求地址、请求方法、请求头和请求体等部分组成：

请求通过 IP 地址、端口和路径定位到具体服务：

3. Ollama 常用命令

# 查看已经下载的模型
ollama list

# 下载模型
ollama pull 模型名称:参数规模

# 运行模型；本地不存在时会先下载
ollama run 模型名称:参数规模

# 复制模型
ollama cp 原模型名称 新模型名称

# 删除模型
ollama rm 模型名称

# 查看当前正在运行的模型
ollama ps

# 查看模型信息
ollama show 模型名称

进入 Ollama 对话界面后，可以使用：

/clear           清除当前会话上下文
/load 模型名称   切换模型
/bye             退出对话

多行输入可以使用三个双引号包裹：

"""
第一行
第二行
第三行
"""

4. 使用 Chatbox 连接本地模型

基本配置流程：

1. 启动 Ollama，并确认模型已经下载。
2. 在 Chatbox 中选择 Ollama API。
3. 配置 API 地址和模型名称。
4. 在聊天界面中测试模型是否能够正常回答。

5. 使用代码搭建聊天界面

可以使用以下组合：

• Streamlit：构建聊天页面。
• Python Ollama SDK：调用本地模型。
• Ollama：运行模型并提供后端服务。

这条路线需要先掌握 Python 基础语法，然后再学习第三方包、API 调用和界面状态管理。

二、Python 基础语法

1. 字面量和常见数据类型

字面量是直接写在代码中的具体值。

类型	示例
整数 int	123
浮点数 float	1.23
复数 complex	1 + 23j
布尔值 bool	True、False
字符串 str	'hello'、"hello"
列表 list	[1, 2, 3]
元组 tuple	(1, 2, 3)
集合 set	{1, 2, 3}
字典 dict	{"name": "小明"}

2. 注释

# 这是单行注释

"""
这是一个多行字符串。
它经常被用作文档字符串，但不等同于真正的注释。
"""

PyCharm 中可以使用 Ctrl + / 快速添加或取消单行注释。

3. PyCharm 常用快捷键

操作	快捷键
注释或取消注释	Ctrl + /
复制	Ctrl + C
粘贴	Ctrl + V
复制当前行	Ctrl + D
剪切	Ctrl + X
撤销	Ctrl + Z
重做	Ctrl + Y
上下移动当前行	Ctrl + Shift + ↑/↓
格式化代码	Ctrl + Alt + L
多光标编辑	Alt + 鼠标左键

具体快捷键可能会受到操作系统和自定义键位设置影响。

4. 变量

变量用于保存对象的引用：

model_name = "qwen"
temperature = 0.7

# 修改变量
temperature = 0.5

# 类型标注用于说明预期类型，默认不会在运行时强制限制
release_year: int = 2025

Python 是动态类型语言：变量不需要提前声明固定类型，同一个变量可以在运行过程中绑定不同类型的对象。

5. 类型查看和转换

value = "123"

print(type(value))
number = int(value)
decimal = float(value)
text = str(number)
flag = bool(value)

常用转换函数：

• type()：查看对象类型。
• int()：转换为整数。
• float()：转换为浮点数。
• str()：转换为字符串。
• bool()：转换为布尔值。

不是所有内容都能转换为数字。例如 int("hello") 会触发 ValueError。

以下值转换为布尔值时通常为 False：

0
0.0
""
[]
{}
set()
None

6. 标识符命名

标识符包括变量名、函数名和类名。

基本规则：

1. 可以包含字母、数字和下划线，但不能以数字开头。
2. 严格区分大小写。
3. 不能使用 Python 关键字。
4. 虽然可以使用中文，但不推荐在正式项目中使用。

推荐规范：

# 变量名和函数名使用小写加下划线
model_name = "qwen"

def load_model():
    pass

# 类名使用大驼峰
class ChatModel:
    pass

7. 分析代码报错

建议从下往上阅读错误信息：

1. 确认发生错误的文件。
2. 确认行号。
3. 找到对应代码。
4. 阅读最后一行的异常类型和原因。
5. 修改后重新运行，确认问题是否解决。

8. 运算符

算术运算符

a + b   # 加法
a - b   # 减法
a * b   # 乘法
a / b   # 除法，结果通常为浮点数
a // b  # 整除
a % b   # 取余
a ** b  # 幂运算

赋值与复合赋值

number = 10
number += 1
number -= 1
number *= 2
number /= 2
number //= 2
number %= 2
number **= 2

比较运算符

a > b
a < b
a >= b
a <= b
a == b
a != b

Python 不会自动把字符串数字和整数转换成相同类型，因此 "1" == 1 的结果是 False。

逻辑运算符

and  # 与
or   # 或
not  # 非

and 和 or 具有短路特性，并且可能返回参与运算的原始值：

print("TO BE" or "NOT TO BE")  # TO BE
print("" or "NOT TO BE")       # NOT TO BE
print("TO BE" and "NOT TO BE") # NOT TO BE
print("" and "NOT TO BE")      # 空字符串

9. 字符串

name = "小明"
age = 18
height = 1.75

# 百分号格式化
print("姓名：%s，年龄：%d，身高：%.2f" % (name, age, height))

# format
print("姓名：{}，年龄：{}".format(name, age))

# 推荐使用 f-string
print(f"姓名：{name}，年龄：{age}，身高：{height:.2f}")

10. 输入和输出

name = input("请输入姓名：")
print("你好", name)
print("这行结尾不换行", end=" ")

input() 会等待用户输入，并且返回值始终是字符串。需要数字时应显式转换：

age = int(input("请输入年龄："))

不建议使用 eval() 解析不可信输入，因为它可能执行任意代码。解析 JSON 时应使用 json.loads()，转换数字时应使用 int() 或 float()。

三、流程控制

1. 分支语句

单分支

age = int(input("请输入年龄："))

if age >= 18:
    print("已经成年")

双分支

if age >= 18:
    print("已经成年")
else:
    print("尚未成年")

多分支

score = 85

if score >= 90:
    level = "优秀"
elif score >= 80:
    level = "良好"
elif score >= 60:
    level = "及格"
else:
    level = "不及格"

嵌套分支

account = input("请输入账号：")
password = input("请输入密码：")

if account == "admin":
    if password == "123456":
        print("登录成功")
    else:
        print("密码错误")
else:
    print("账号不存在")

这个例子也可以使用逻辑运算符简化：

if account == "admin" and password == "123456":
    print("登录成功")
else:
    print("账号或密码错误")

2. 数字猜谜案例

import random

answer = random.randint(1, 10)

for attempt in range(3):
    guess = int(input("请输入 1～10 之间的数字："))

    if guess == answer:
        print("猜对了")
        break
    if guess > answer:
        print("猜大了")
    else:
        print("猜小了")
else:
    print(f"机会用完了，正确答案是 {answer}")

3. while 循环

while 会在条件为真时重复执行代码：

count = 1

while count <= 10:
    print(f"第 {count} 次执行")
    count += 1

循环的三个常见组成部分：

1. 定义状态变量。
2. 设置继续或终止条件。
3. 在循环体中更新状态。

需要持续接收用户操作时，可以使用 while True，并通过 break 退出：

while True:
    choice = input("请输入操作，输入 q 退出：")
    if choice == "q":
        break
    print(f"正在执行：{choice}")

4. for 循环

for 适合依次处理可迭代对象中的每一项：

for model_name in ["qwen", "llama", "deepseek"]:
    print(model_name)

使用 range() 控制循环次数：

total = 0

for number in range(1, 101):
    total += number

print(total)

5. 循环嵌套：九九乘法表

for row in range(1, 10):
    for column in range(1, row + 1):
        print(f"{column}×{row}={column * row}", end="\t")
    print()

四、Python 常用容器

1. 容器对比

容器	示例	是否有序	是否可变	是否允许重复
字符串	"abc"	是	否	是
列表	[1, 2, 3]	是	是	是
元组	(1, 2, 3)	是	否	是
集合	{1, 2, 3}	否	是	否
字典	{"name": "小明"}	保留插入顺序	是	键不能重复

2. 索引和切片

字符串、列表和元组都支持索引与切片：

text = "transformer"

print(text[0])     # t
print(text[-1])    # r
print(text[0:5])   # trans
print(text[::2])   # 每隔一个字符取一次

切片遵循“包含起始位置，不包含结束位置”的规则。

3. 字符串常用方法

text = "  hello python  "

len(text)
text.count("o")
text.index("python")
text.strip()
text.split(" ")
text.replace("python", "llm")

字符串不可变，这些方法通常返回新的字符串，不会直接修改原字符串。

4. 列表

models = ["qwen", "llama"]

# 新增
models.append("deepseek")
models.insert(1, "gemma")
models.extend(["phi", "mistral"])

# 删除
models.remove("phi")
removed = models.pop()
del models[0]

# 修改
models[0] = "qwen2"

# 查询
len(models)
models.count("qwen2")
models.index("qwen2")

# 排序
numbers = [3, 1, 2]
numbers.sort()
numbers.sort(reverse=True)

5. 元组

point = (10, 20)
single = (1,)  # 单元素元组必须保留逗号

len(point)
point.count(10)
point.index(20)

元组不可变，适合保存不希望被修改的数据。

6. 集合

skills = {"python", "rag", "agent"}

skills.add("pytorch")
skills.remove("rag")

a = {1, 2, 3}
b = {3, 4, 5}

print(a | b)  # 并集
print(a & b)  # 交集
print(a - b)  # 差集

集合无序且元素唯一，适合去重和集合运算。

7. 字典

model = {
    "name": "qwen",
    "temperature": 0.7,
}

# 新增或修改
model["year"] = 2025
model["temperature"] = 0.5

# 查询
print(model["name"])
print(model.get("provider", "未知"))
print(model.keys())
print(model.values())

# 删除
model.pop("year")

使用字典统计字符出现次数：

text = "hello"
counter = {}

for char in text:
    counter[char] = counter.get(char, 0) + 1

print(counter)

8. 容器类型转换

list((1, 2, 3))
tuple([1, 2, 3])
set([1, 2, 2, 3])
list({1, 2, 3})

列表和元组可以使用 + 拼接，集合应使用并集等集合运算。

9. 推导式和条件表达式

# 列表推导式
squares = [number ** 2 for number in range(5)]

# 集合推导式
unique_lengths = {len(word) for word in ["llm", "rag", "agent"]}

# 字典推导式
square_map = {number: number ** 2 for number in range(5)}

# 条件表达式
status = "成年" if age >= 18 else "未成年"

五、函数

1. 定义和调用函数

def greet():
    print("你好")

greet()

按照参数和返回值可以分为四种形式：

# 无参数、无返回值
def show_message():
    print("hello")

# 有参数、无返回值
def show_model(model_name):
    print(model_name)

# 无参数、有返回值
def get_default_model():
    return "qwen"

# 有参数、有返回值
def get_sum(start, end):
    return sum(range(start, end + 1))

2. 函数参数

def generate_text(prompt, model="qwen", temperature=0.7):
    return f"模型={model}，温度={temperature}，提示词={prompt}"

# 位置参数
generate_text("你好", "llama", 0.5)

# 关键字参数
generate_text(prompt="你好", temperature=0.2)

不定长参数：

def collect_models(*args, **kwargs):
    print(args)    # 位置参数组成的元组
    print(kwargs)  # 关键字参数组成的字典

3. 文档字符串

def add(a, b):
    """返回两个数的和。"""
    return a + b

4. 变量作用域

• 在函数外定义的变量通常属于全局作用域。
• 在函数内定义的变量通常属于局部作用域。
• 函数外不能直接访问函数内部的局部变量。
• 虽然可以使用 global 修改全局变量，但一般应优先通过参数和返回值传递数据。

5. 多返回值和解包

def get_user():
    return "小明", 18

name, age = get_user()

实际上，多个返回值会被组织成一个元组。

6. Lambda 表达式

Lambda 适合编写简单的匿名函数：

add = lambda x, y: x + y
result = add(1, 2)

复杂逻辑应优先使用普通的 def 函数，以提高可读性。

7. 递归

递归是函数调用自身。一个正确的递归通常需要：

1. 问题可以被拆成规模更小的同类问题。
2. 存在明确的终止条件。
3. 正确返回每一层的计算结果。

计算累加和：

def recursive_sum(number):
    if number == 1:
        return 1
    return number + recursive_sum(number - 1)

print(recursive_sum(10))

计算阶乘：

def factorial(number):
    if number <= 1:
        return 1
    return number * factorial(number - 1)

计算斐波那契数列：

def fibonacci(number):
    if number <= 1:
        return number
    return fibonacci(number - 1) + fibonacci(number - 2)

这个递归版本在输入较大时会重复计算，实际项目中可以使用循环、缓存或动态规划进行优化。

递归统计嵌套字典中的键：

def count_keys(data):
    total = len(data)

    for value in data.values():
        if isinstance(value, dict):
            total += count_keys(value)

    return total

递归统计大模型研发团队的成员总数。团队下面既可以直接记录人数，也可以继续划分为更小的项目组，因此需要逐层遍历：

ai_team = {
    "模型研发组": {
        "数据处理小组": 4,
        "训练优化小组": 6,
        "模型评测小组": 3,
    },
    "应用开发组": {
        "知识库项目": {
            "检索模块": 2,
            "生成模块": 3,
        },
        "智能体项目": 4,
    },
    "平台运维组": 2,
}


def count_members(team):
    """统计任意层级团队结构中的成员总数。"""
    member_total = 0

    for group in team.values():
        if isinstance(group, int):
            member_total += group
        elif isinstance(group, dict):
            member_total += count_members(group)

    return member_total


total = count_members(ai_team)
print(f"研发团队共有 {total} 人")

六、异常、文件和系统操作

1. 异常处理

异常处理可以防止预期内的错误直接中断整个程序：

try:
    number = int(input("请输入整数："))
except ValueError as error:
    print("输入内容不是有效整数", error)
else:
    print("转换成功", number)
finally:
    print("处理结束")

应尽量捕获具体异常，例如 ValueError、FileNotFoundError，避免无条件忽略所有错误。

2. 文件读写

常用模式：

模式	说明
r	只读；文件必须存在
w	写入；不存在时创建，存在时覆盖
a	追加；不存在时创建
b	二进制模式，可与其他模式组合
+	读写模式，可与其他模式组合

推荐使用 with，操作完成后文件会自动关闭：

with open("notes.txt", mode="r", encoding="utf-8") as file:
    content = file.read()

with open("result.txt", mode="w", encoding="utf-8") as file:
    file.write("学习记录")

逐行读取：

with open("notes.txt", encoding="utf-8") as file:
    for line in file:
        print(line.rstrip())

常见路径形式：

• 绝对路径：从盘符或根目录开始的完整路径。
• 相对路径：相对于当前工作目录，例如 ../data/file.txt。

3. os 模块

import os

os.mkdir("data")       # 创建目录
os.rmdir("data")       # 删除空目录
os.remove("file.txt")  # 删除文件
os.listdir(".")        # 查看目录内容
os.getenv("PATH")      # 读取环境变量

实际项目也可以使用接口更现代的 pathlib 模块处理路径。

七、Linux 常用命令

1. 查看和切换目录

ls          # 查看当前目录内容
ls -a       # 显示隐藏文件
ls -l       # 显示详细属性
ls -lh      # 使用易读单位显示文件大小
cd 路径     # 切换目录
cd ..       # 返回上一级
pwd         # 查看当前路径
clear       # 清屏

2. 帮助

命令 --help
man 命令

3. 文件和目录

mkdir 目录名          # 创建目录
touch 文件名          # 创建空文件或更新时间戳
cat -n 文件名         # 查看较短文件并显示行号
more 文件名           # 分页查看文件
head 文件名           # 查看文件开头
tail 文件名           # 查看文件结尾
cp 源文件 目标路径    # 复制文件
cp -r 源目录 目标路径 # 复制目录
mv 原路径 新路径      # 移动或重命名
rm 文件名             # 删除文件
rm -r 目录名          # 递归删除目录，使用前要确认路径

重定向：

echo "内容" > 文件名   # 覆盖写入
echo "内容" >> 文件名  # 追加写入

4. 打包和解包

tar -cvf archive.tar 文件或目录
tar -xvf archive.tar
tar -zcvf archive.tar.gz 文件或目录
tar -zxvf archive.tar.gz

5. Vim 基本操作

vim 文件名   打开文件
i            从普通模式进入插入模式
Esc          返回普通模式
:wq          保存并退出
:q           退出未修改的文件
:q!          放弃修改并退出

八、面向对象

面向对象编程把相关的数据和行为组织在一个类中。类负责描述一类对象的共同结构，对象是根据类创建出来的具体实例。

1. 定义类和创建对象

Python 3 中通常直接使用 class 类名:：

class ChatModel:
    provider = "本地服务"  # 类属性

    def reply(self, prompt):
        return f"收到问题：{prompt}"


model = ChatModel()
print(model.provider)
print(model.reply("什么是 Transformer？"))

在实例方法中，self 指向调用该方法的实例。通过 self.属性名 可以访问实例自己的状态。

2. 使用 __init__ 初始化对象

__init__() 会在创建实例时自动调用，适合设置每个对象各自拥有的属性：

class ChatModel:
    def __init__(self, name, temperature=0.7):
        self.name = name
        self.temperature = temperature

    def describe(self):
        return f"模型：{self.name}，温度：{self.temperature}"


qwen = ChatModel("qwen", 0.3)
llama = ChatModel("llama", 0.8)

print(qwen.describe())
print(llama.describe())

与创建对象后再手动添加属性相比，统一在 __init__() 中初始化可以保证对象一创建就处于完整状态。

3. 常见特殊方法

特殊方法也常被称为魔术方法，会在特定操作发生时由 Python 调用：

class ModelConfig:
    def __init__(self, name, context_length):
        self.name = name
        self.context_length = context_length

    def __str__(self):
        return f"{self.name}（上下文长度：{self.context_length}）"


config = ModelConfig("qwen", 32768)
print(config)  # 自动调用 __str__

常见特殊方法：

• __init__()：初始化实例。
• __str__()：定义 print(对象) 和 str(对象) 的显示内容。
• __len__()：定义 len(对象) 的结果。
• __enter__()、__exit__()：支持 with 上下文管理。
• __del__()：对象回收前可能被调用，但执行时机不应作为业务逻辑依据。

4. 封装

封装是把数据和操作数据的方法放在一起，并通过清晰的接口限制不合理操作：

class ModelQuota:
    def __init__(self, total_tokens):
        self.__remaining_tokens = total_tokens

    @property
    def remaining_tokens(self):
        return self.__remaining_tokens

    def consume(self, amount):
        if amount <= 0:
            raise ValueError("消耗量必须大于 0")
        if amount > self.__remaining_tokens:
            raise ValueError("剩余 Token 不足")
        self.__remaining_tokens -= amount


quota = ModelQuota(10_000)
quota.consume(800)
print(quota.remaining_tokens)

双下划线属性会触发名称改写，用于降低类外部误用的概率，但它不是绝对的安全边界。

5. 继承、方法重写和 super()

继承可以复用父类已有的属性和方法，子类也可以重写其中的行为：

class BaseModel:
    def __init__(self, name):
        self.name = name

    def generate(self, prompt):
        return f"{self.name} 正在处理：{prompt}"


class LocalModel(BaseModel):
    def __init__(self, name, host):
        super().__init__(name)
        self.host = host

    def generate(self, prompt):
        base_message = super().generate(prompt)
        return f"{base_message}，服务地址：{self.host}"


model = LocalModel("qwen", "http://127.0.0.1:11434")
print(model.generate("介绍一下 RAG"))

super() 会按照方法解析顺序 MRO 寻找下一个实现，可以通过 类名.mro() 查看解析顺序。多继承场景中，使用 super() 通常比直接调用某个父类的方法更容易维护。

6. 多态

不同类型的对象只要提供相同的方法，就可以通过同一段代码使用：

class LocalModel:
    def generate(self, prompt):
        return f"本地模型回答：{prompt}"


class CloudModel:
    def generate(self, prompt):
        return f"云端模型回答：{prompt}"


def ask(model, prompt):
    print(model.generate(prompt))


ask(LocalModel(), "你好")
ask(CloudModel(), "你好")

Python 更强调对象是否提供所需行为，而不要求它们必须继承同一个父类，这种方式常被称为鸭子类型。

7. 类方法和静态方法

class ModelConfig:
    default_temperature = 0.7

    def __init__(self, name, temperature):
        self.name = name
        self.temperature = temperature

    @classmethod
    def create_default(cls, name):
        return cls(name, cls.default_temperature)

    @staticmethod
    def is_valid_temperature(value):
        return 0 <= value <= 2

• 实例方法的第一个参数通常是 self，主要操作实例状态。
• 类方法使用 @classmethod，第一个参数通常是 cls，主要操作类级别状态或作为替代构造器。
• 静态方法使用 @staticmethod，不自动接收实例或类，适合放置与该类密切相关的辅助逻辑。

九、闭包和装饰器

1. 闭包

当内部函数引用了外部函数的变量，并且外部函数返回内部函数时，内部函数会保留这些变量：

def create_counter(start=0):
    count = start

    def increase():
        nonlocal count
        count += 1
        return count

    return increase


counter = create_counter(10)
print(counter())  # 11
print(counter())  # 12

闭包常用于保存少量状态、创建函数工厂，以及实现装饰器。只要闭包仍然被引用，它需要使用的外部变量也会继续存在。

2. 装饰器

装饰器在不直接修改原函数代码的情况下，为函数增加额外行为：

from functools import wraps
from time import perf_counter


def measure_time(function):
    @wraps(function)
    def wrapper(*args, **kwargs):
        start = perf_counter()
        result = function(*args, **kwargs)
        elapsed = perf_counter() - start
        print(f"{function.__name__} 用时 {elapsed:.4f} 秒")
        return result

    return wrapper


@measure_time
def generate_answer(prompt):
    return f"回答：{prompt}"


print(generate_answer("什么是向量数据库？"))

@measure_time 等价于：

generate_answer = measure_time(generate_answer)

使用 *args 和 **kwargs 可以让包装函数接收不同形式的参数；保留并返回原函数的结果，可以兼容有无返回值的函数。functools.wraps 用于保留原函数的名称和文档等元数据。

3. 带参数的装饰器

如果装饰器本身也需要配置，可以再增加一层函数：

from functools import wraps


def retry(max_attempts):
    def decorator(function):
        @wraps(function)
        def wrapper(*args, **kwargs):
            last_error = None

            for _ in range(max_attempts):
                try:
                    return function(*args, **kwargs)
                except ConnectionError as error:
                    last_error = error

            raise last_error

        return wrapper

    return decorator

多个装饰器会从靠近函数的一层开始包装，调用时则从最外层进入。

十、网络编程

1. TCP、IP 和端口

• IP 地址用于定位网络中的主机或网络接口。
• 端口用于定位主机上的具体网络服务。
• TCP 是面向连接、可靠、有序的字节流协议。
• UDP 是无连接的数据报协议，不保证送达、顺序或去重。

TCP 只传输字节，不保留应用层消息边界。因此发送文本时需要编码，接收后需要解码；传输多段或大文件时，还应自行设计消息长度、分隔符或其他应用层协议。

2. TCP 客户端

import socket


with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
    client.connect(("127.0.0.1", 10086))

    message = input("请输入聊天内容：")
    client.sendall(message.encode("utf-8"))

    response = client.recv(1024)
    print(response.decode("utf-8"))

客户端的基本步骤是创建套接字、连接服务器、发送数据、接收响应并关闭连接。

3. TCP 服务端

import socket


with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(("127.0.0.1", 10086))
    server.listen(10)

    print("服务器正在监听 127.0.0.1:10086")

    while True:
        connection, address = server.accept()

        with connection:
            request = connection.recv(1024)
            print(address, request.decode("utf-8"))
            connection.sendall("消息已收到".encode("utf-8"))

服务端的基本步骤是创建套接字、绑定 IP 和端口、开始监听、接收连接、收发数据并关闭本次连接。

4. UDP 客户端

UDP 不需要提前建立连接：

import socket


with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client:
    message = "这是一条 UDP 消息".encode("utf-8")
    client.sendto(message, ("127.0.0.1", 10087))

UDP 适合对实时性要求较高、能够容忍少量丢包，或者由应用层自行处理可靠性的场景。

5. TCP 服务中转到 Ollama

TCP 服务端可以接收用户问题，再把问题转交给 Ollama：

import socket

import ollama


ollama_client = ollama.Client(host="http://127.0.0.1:11434")

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(("127.0.0.1", 10086))
    server.listen(5)

    connection, address = server.accept()

    with connection:
        prompt = connection.recv(4096).decode("utf-8")
        response = ollama_client.chat(
            model="模型名称",
            messages=[{"role": "user", "content": prompt}],
        )
        answer = response.message.content
        connection.sendall(answer.encode("utf-8"))

这个示例只处理一次连接和一条短消息，适合帮助理解数据流。正式服务还需要考虑并发、超时、身份认证、异常处理、消息边界和输入长度限制。

6. 通过 TCP 传输文件

单次 recv() 不保证得到完整文件。一个简单做法是先发送固定长度的文件大小，再循环接收文件内容：

客户端：

import socket
import struct
from pathlib import Path


file_path = Path("image.png")
file_size = file_path.stat().st_size

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
    client.connect(("127.0.0.1", 10088))
    client.sendall(struct.pack("!Q", file_size))

    with file_path.open("rb") as file:
        while chunk := file.read(64 * 1024):
            client.sendall(chunk)

服务端：

import socket
import struct
from pathlib import Path


def receive_exactly(connection, size):
    chunks = []
    received = 0

    while received < size:
        chunk = connection.recv(size - received)
        if not chunk:
            raise ConnectionError("连接提前中断")
        chunks.append(chunk)
        received += len(chunk)

    return b"".join(chunks)


output_path = Path("received-image.png")

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.bind(("127.0.0.1", 10088))
    server.listen(5)
    connection, _ = server.accept()

    with connection:
        file_size = struct.unpack("!Q", receive_exactly(connection, 8))[0]

        with output_path.open("wb") as file:
            remaining = file_size
            while remaining:
                chunk = connection.recv(min(64 * 1024, remaining))
                if not chunk:
                    raise ConnectionError("文件接收不完整")
                file.write(chunk)
                remaining -= len(chunk)

实际服务还应校验文件名、文件类型和最大文件大小，避免路径穿越和超大文件占满磁盘。

十一、进程、线程与协程

1. 基本区别

方式	主要特点	常见用途
进程	内存空间相对独立，创建和通信成本较高	CPU 密集型计算、任务隔离
线程	同一进程内共享内存，切换成本较低	阻塞式 I/O、多个连接
协程	在单线程内协作式调度，需要显式等待	大量网络请求、异步 I/O

进程是操作系统进行资源分配和隔离的重要单位，线程是操作系统调度执行的重要单位。具体实现会受到操作系统和运行时影响。

2. 多进程

import multiprocessing
import time


def run_task(name):
    for number in range(3):
        print(name, number)
        time.sleep(0.5)


if __name__ == "__main__":
    process_a = multiprocessing.Process(
        target=run_task,
        args=("任务 A",),
    )
    process_b = multiprocessing.Process(
        target=run_task,
        args=("任务 B",),
    )

    process_a.start()
    process_b.start()
    process_a.join()
    process_b.join()

在 Windows 中创建子进程时，应把启动代码放在 if __name__ == "__main__": 保护块中。普通全局变量不会自动在进程间共享；需要通信时可以使用队列、管道、共享内存或进程管理器。

3. 多线程

import threading
import time


def download(name):
    print(f"开始下载 {name}")
    time.sleep(1)
    print(f"完成下载 {name}")


threads = [
    threading.Thread(target=download, args=(f"文件 {index}",))
    for index in range(3)
]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

线程共享同一进程的内存，多个线程同时修改共享数据时可能发生竞争条件。

4. 互斥锁

import threading
import time


remaining_tickets = 3
lock = threading.Lock()


def buy_ticket():
    global remaining_tickets

    with lock:
        if remaining_tickets <= 0:
            print(f"{threading.current_thread().name}：票已售罄")
            return

        time.sleep(0.2)
        remaining_tickets -= 1
        print(
            f"{threading.current_thread().name} 购买成功，"
            f"剩余 {remaining_tickets} 张"
        )


threads = [
    threading.Thread(target=buy_ticket, name=f"用户 {index}")
    for index in range(1, 6)
]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

锁只应包住必须同步的临界区。锁的范围过大会降低并发效率，多个锁使用不当还可能产生死锁。

5. GIL 的影响

CPython 的全局解释器锁 GIL 会限制同一进程中多个线程同时执行 Python 字节码。因此：

• CPU 密集型纯 Python 任务通常更适合多进程或使用能释放 GIL 的底层计算库。
• I/O 密集型任务在等待网络或磁盘时会释放执行机会，多线程仍然很有价值。
• GIL 不等于线程安全，共享可变数据仍然可能需要锁或线程安全的数据结构。

6. 守护任务

守护线程或守护进程会在非守护主任务结束后被运行时终止，不适合承担必须完成的保存、写盘或事务提交工作。正常退出时应优先使用事件、队列等方式通知任务结束，并调用 join() 等待清理完成。

十二、生成器、协程和网络请求

1. 生成器表达式

列表推导式会立即生成完整列表，生成器表达式则按需产生数据：

numbers = [number * 2 for number in range(10)]
number_generator = (number * 2 for number in range(10))

print(next(number_generator))
print(next(number_generator))

for number in number_generator:
    print(number)

生成器不会一次把全部结果保存在内存中，因此特别适合处理大批量或流式数据。它仍会占用少量状态内存，并非完全“不占内存”。

2. 使用 yield 创建生成器

def read_batches(items, batch_size):
    for start in range(0, len(items), batch_size):
        yield items[start : start + batch_size]


for batch in read_batches(list(range(10)), 3):
    print(batch)

调用包含 yield 的函数时不会立即执行完整函数，而是返回生成器对象。每次迭代会运行到下一个 yield，并保留当前执行状态。

3. 协程

async def 定义协程函数，await 用于等待支持异步协议的操作：

import asyncio


async def run_task(name, delay):
    await asyncio.sleep(delay)
    return f"{name} 完成"


async def main():
    tasks = [
        asyncio.create_task(run_task("任务 A", 1)),
        asyncio.create_task(run_task("任务 B", 1)),
    ]
    results = await asyncio.gather(*tasks)
    print(results)


asyncio.run(main())

协程适合大量需要等待的 I/O 任务。计算密集型代码如果长时间不执行 await，仍然会阻塞事件循环。

4. 使用 HTTPX 发送请求

同步请求：

import httpx


with httpx.Client(timeout=10) as client:
    response = client.get("https://example.com/api/items")
    response.raise_for_status()
    data = response.json()

异步请求：

import asyncio

import httpx


async def fetch(client, url):
    response = await client.get(url)
    response.raise_for_status()
    return response.json()


async def main():
    urls = [
        "https://example.com/api/items/1",
        "https://example.com/api/items/2",
    ]

    async with httpx.AsyncClient(timeout=10) as client:
        results = await asyncio.gather(
            *(fetch(client, url) for url in urls)
        )

    print(results)


asyncio.run(main())

example.com 是占位地址，运行时应替换为真实且允许访问的 API。

5. 下载并保存二进制文件

from pathlib import Path

import httpx


url = "https://example.com/files/image.png"
output = Path("files/image.png")
output.parent.mkdir(parents=True, exist_ok=True)

with httpx.Client(timeout=30, follow_redirects=True) as client:
    response = client.get(url)
    response.raise_for_status()
    output.write_bytes(response.content)

批量抓取网页或图片前，需要确认站点条款、版权要求、robots.txt 和访问频率限制。不要通过伪造请求头绕过访问控制，也不要保存或传播无权使用的内容。

十三、正则表达式

正则表达式适合匹配具有明确文本模式的内容。

1. 常用匹配函数

import re


text = "model=qwen; tokens=4096"

re.match(r"model", text)          # 从字符串开头匹配
re.search(r"tokens=\d+", text)   # 查找第一个匹配
re.findall(r"\d+", text)         # 返回全部匹配结果

2. 常用语法

语法	含义
.	除换行符外的任意字符
[0-9a-zA-Z]	字符集合或范围
\d	数字字符
\s	空白字符
\w	字母、数字和下划线等单词字符
*	前一项出现 0 次或多次
+	前一项出现 1 次或多次
?	前一项出现 0 次或 1 次
{5}	固定出现 5 次
{3,5}	出现 3～5 次
{5,}	至少出现 5 次
^	字符串或行开头
$	字符串或行结尾
|	或
(...)	捕获分组

3. 贪婪和非贪婪匹配

量词默认会尽可能多地匹配，增加 ? 后会尽可能少地匹配：

html = '<img src="first.png"><img src="second.png">'

greedy = re.findall(r'src="(.*)"', html)
non_greedy = re.findall(r'src="(.*?)"', html)

print(greedy)
print(non_greedy)

4. 分组和反向引用

text = "<h1>标题</h1>"
result = re.fullmatch(r"<(h[1-6])>.*</\1>", text)

(h[1-6]) 是第一个捕获分组，\1 表示再次匹配该分组已经匹配到的内容。

5. 输入格式校验

def is_valid_model_name(name):
    pattern = r"^[a-zA-Z0-9_-]+(?::[a-zA-Z0-9_.-]+)?$"
    return re.fullmatch(pattern, name) is not None


print(is_valid_model_name("qwen:7b"))
print(is_valid_model_name("model name"))

正则适合处理文本模式，但不适合完整解析复杂 HTML。处理网页结构时，通常应使用专门的 HTML 解析器。

十四、后续学习方向

完成这些基础内容后，可以继续学习：

1. 使用 Python 调用 Ollama API。
2. 使用 Streamlit 构建聊天界面。
3. 学习 PyTorch 张量和神经网络基础。
4. 学习 Transformers 模型加载与推理。
5. 构建第一个 RAG 知识库。
6. 记录模型参数、运行环境和实验结果。

这份笔记的目标不是一次记住所有 API，而是在后续项目中遇到问题时能够快速定位相关知识，并继续补充新的理解。

返回笔记开头 ↑