前提: 构建菜谱知识库 (参照例子)
向量数据库:Qdrant (混合搜索: 稠密向量 + 稀疏向量 + 融合算法)
大模型llm: deepseek
embedding模型: 智谱embedding-3 (官方文档)

搭建向量数据库:
1、创建集合

# 创建集合
await client.create_collection(
    collection_name=collection_name,
    vectors_config={
        # Dense vectors for semantic understanding of research concepts
        # "dense": models.VectorParams(size=384, distance=models.Distance.COSINE),
        "dense": models.VectorParams(size=512, distance=models.Distance.COSINE),
        # ColBERT multivectors for fine-grained text understanding
        # "colbert": models.VectorParams(
        #     size=128,
        #     distance=models.Distance.COSINE,
        #     multivector_config=models.MultiVectorConfig(
        #         comparator=models.MultiVectorComparator.MAX_SIM
        #     ),
        # ),
    },
    sparse_vectors_config={
        # Sparse vectors for exact technical term matching
        "sparse": models.SparseVectorParams(
            modifier=models.Modifier.IDF,
            index=models.SparseIndexParams(
                # full_scan_threshold=0, # 在达到此数值(不包括此数值)之前,比较期间 不会 使用倒排索引(但 仍会构建 索引)。
                on_disk=False, # 将倒排索引存储在磁盘上 (True) 还是内存中 (False,默认值)。
            )
        )
    },
)

2、上传数据

foods = [{...}...]
texts = [it["cooking_instructions"] for it in foods]

points = []
for i, text in enumerate(texts):

    denceResults = await get_embedding_zhipu(text, 512)

    points.append(
        models.PointStruct(
            id=i,
            vector={
                # "dense": models.Document(
                #     text=text,
                #     model="sentence-transformers/all-MiniLM-L6-v2",
                # ),
                "dense": denceResults[0],
                "sparse": models.Document(
                    text=text,
                    model="Qdrant/bm25",
                ),
            },
            payload=foods[i],
        )
    )

client.upload_points(
    collection_name=collection_name,
    points=points,
)

3、查询

async def rrf_search(query: str) -> list[models.ScoredPoint]:
    denceQuerys = await get_embedding_zhipu(query, 512)

    response = await client.query_points(
        collection_name=collection_name,
        prefetch=[
            models.Prefetch(
                query=denceQuerys[0],
                using="dense",
                limit=5,
            ),
            models.Prefetch(
                query=models.Document(
                    text=query,
                    model="Qdrant/bm25",
                ),
                using="sparse",
                limit=5,
            ),
        ],
        query=models.FusionQuery(fusion=models.Fusion.RRF),
        limit=5,
    )
    return response.points

搜索流程:
step1、用户输入: 有没有蔬菜类的汤羹,不包含肉类的,给我推荐一下
step2、根据用户输入去向量知识库进行查询,返回相近结果
step3、根据相近结果,用大模型llm再进行一次判断识别,获取较为合适的结果。

例子:
有没有蔬菜类的汤羹,不包含肉类的,给我推荐一下
请根据以下选项挑选出符合要求的选项,并根据对应选项内容进行回答,回答要自然得体,不要显示“符合要求”等相关字眼,不要有其他不符合要求的选项的说明。

  • 番茄豆腐汤 1.番茄切块,豆腐切小方块;2.热油炒番茄至出汁;3.加清水烧开,下豆腐煮5分钟;4.淋蛋液搅出蛋花;5.加盐调味,撒葱花淋香油即可。风味酸甜鲜香、豆腐嫩滑、汤汁清爽;低脂素食,适合减脂人群;适宜夏季开胃;适合老人、儿童及清淡饮食者。 0.5
  • 上汤娃娃菜 1.娃娃菜洗净纵向切条;2.热油爆香蒜末,加清水烧开;3.下皮蛋丁、咸鸭蛋丁煮出鲜味;4.下娃娃菜煮3分钟至软;5.加盐调味,撒枸杞出锅。风味汤鲜味美、娃娃菜软嫩、咸鲜适口;含蛋类,过敏者忌食;适宜四季,秋冬尤佳;适合老人、儿童及清淡饮食人群。 0.33333334
  • 冬瓜排骨汤 1.排骨冷水下锅加料酒焯水去浮沫;2.排骨加姜片、葱段和清水,大火烧开转小火炖40分钟;3.冬瓜去皮切块下锅再炖15分钟;4.加盐调味,撒枸杞即可。风味汤清味鲜、冬瓜软糯、排骨酥烂;清淡少油,适合老人儿童;适宜夏秋季节清热利水;适合体虚调养及家庭日常人群。 0.25
  • 鲫鱼豆腐汤 1.鲫鱼处理干净,两面划刀;2.热油下鲫鱼煎至两面金黄;3.加姜片、料酒和开水大火煮至汤白;4.下豆腐块煮10分钟;5.加盐、白胡椒粉调味,撒葱花香菜即可。风味汤白鲜浓、鱼肉嫩滑、豆腐软嫩;菜品分类:汤羹;含鱼类,过敏者忌食;适宜秋冬季节滋补;适合孕产妇、老人及病后调养人群。 0.2
  • 山药排骨汤 1.排骨冷水下锅加料酒焯水;2.排骨加姜片和清水大火烧开转小火炖40分钟;3.山药去皮切段,玉米切块下锅再炖20分钟;4.加盐调味,撒枸杞即可。风味汤清味鲜、山药软糯、排骨酥烂;温和滋补,适合大众;适宜秋冬春季;适合老人、儿童及脾胃虚弱者。 0.16666667

影响搜索召回率:
1、生成向量的维度尽量大于512, 实测384与512的搜索结果差别大,明显维度越大精度越高
2、文本描述内容组织最好包含一些关键字,精度越好

方法优势权衡最适合
固定大小简单,块可预测忽略结构,破坏含义原始或非结构化文本
基于句子保留完整的思想块大小不一致RAG,问答系统
基于段落与语义单元对齐长度差异大文档、手册、教学内容
滑动窗口保持完整上下文冗余,计算量大重排序,高召回检索
递归灵活,处理混乱输入启发式,有时易碎抓取网页,混合来源
语义高质量,语义感知较慢,资源密集型法律、研究、关键问答

按句子分块实现:

# ! pip install nltk
from nltk.tokenize import sent_tokenize
def sentence_chunk(text, max_words=150):
        sentences = sent_tokenize(text)
        chunks, buffer, length = [], [], 0

        for sent in sentences:
            count = len(sent.split())
            if length + count > max_words:
                chunks.append(" ".join(buffer))
                buffer, length = [], 0
            buffer.append(sent)
            length += count

        if buffer:
            chunks.append(" ".join(buffer))
        return chunks


rocky linux9 安装php + nginx
sudo dnf groupinstall -y "Development Tools"
sudo dnf install -y autoconf automake libtool pkgconf bison re2c

sudo dnf config-manager --set-enabled crb
dnf install -y openssl-devel libxml2-devel curl-devel libpng-devel libjpeg-turbo-devel freetype-devel libzip-devel oniguruma-devel sqlite-devel

./configure \
CFLAGS="-fPIE" \
LDFLAGS="-pie" \
--prefix=/usr/local/php84 \
--with-config-file-path=/usr/local/php84/etc \
--enable-fpm \
--with-fpm-user=www \
--with-fpm-group=www \
--with-mysqli=mysqlnd \
--with-pdo-mysql=mysqlnd \
--with-openssl \
--with-zlib \
--with-curl \
--enable-gd \
--with-jpeg \
--with-freetype \
--enable-mbstring \
--enable-bcmath \
--enable-sockets \
--enable-opcache \
--with-zip \
--with-libxml
--with-sqlite3 \
--with-pdo-sqlite

make -j$(nproc)
make install

sudo cp php.ini-production /usr/local/php84/etc/php.ini
sudo cp /usr/local/php84/etc/php-fpm.conf.default /usr/local/php84/etc/php-fpm.conf
sudo cp /usr/local/php84/etc/php-fpm.d/www.conf.default /usr/local/php84/etc/php-fpm.d/www.conf

sudo tee /etc/systemd/system/php84-fpm.service > /dev/null <<'EOF'
[Unit]
Description=The PHP 8.4 FastCGI Process Manager
After=network.target

[Service]
Type=simple
PIDFile=/usr/local/php84/var/run/php-fpm.pid
ExecStart=/usr/local/php84/sbin/php-fpm --nodaemonize --fpm-config /usr/local/php84/etc/php-fpm.conf
ExecReload=/bin/kill -USR2 $MAINPID
PrivateTmp=true

[Install]
WantedBy=multi-user.target
EOF

echo 'export PATH=/usr/local/php84/bin:/usr/local/php84/sbin:$PATH' | sudo tee /etc/profile.d/php84.sh
source /etc/profile.d/php84.sh

useradd -r -s /sbin/nologin -M www

sudo systemctl daemon-reload
sudo systemctl start php84-fpm
sudo systemctl enable php84-fpm
php -v

安装nginx
dnf install nginx
systemctl enable --now nginx

1、基本模型

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    age: int
    email: str

# 自动类型转换
user = User(id="1", name="Alice", age="25", email="alice@example.com")
print(user)
# id=1 name='Alice' age=25 email='alice@example.com'

print(user.id, type(user.id))  # 1 <class 'int'>

2、字段校验与默认值

from pydantic import BaseModel, Field
from typing import Optional

class Product(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0, description="价格必须大于0")
    quantity: int = Field(default=0, ge=0)
    description: Optional[str] = None

p = Product(name="苹果", price=5.5)
print(p)
# name='苹果' price=5.5 quantity=0 description=None

# 校验失败会抛出 ValidationError
try:
    Product(name="", price=-1)
except Exception as e:
    print(e)

3、嵌套模型

from pydantic import BaseModel
from typing import List

class Address(BaseModel):
    city: str
    street: str

class Person(BaseModel):
    name: str
    age: int
    addresses: List[Address]

data = {
    "name": "Bob",
    "age": 30,
    "addresses": [
        {"city": "北京", "street": "长安街"},
        {"city": "上海", "street": "南京路"},
    ]
}

person = Person(**data)
print(person.addresses[0].city)  # 北京
print(person.model_dump())       # 转 dict
print(person.model_dump_json())  # 转 JSON 字符串

4、自定义校验器

from pydantic import BaseModel, field_validator, model_validator

class User(BaseModel):
    username: str
    password: str
    confirm_password: str

    @field_validator("username")
    @classmethod
    def username_alphanumeric(cls, v: str) -> str:
        if not v.isalnum():
            raise ValueError("用户名必须为字母或数字")
        return v.lower()

    @model_validator(mode="after")
    def check_passwords_match(self):
        if self.password != self.confirm_password:
            raise ValueError("两次密码不一致")
        return self

u = User(username="Alice123", password="abc", confirm_password="abc")
print(u.username)  # alice123

5、枚举与leteral

from pydantic import BaseModel
from enum import Enum
from typing import Literal

class Status(str, Enum):
    active = "active"
    inactive = "inactive"

class Account(BaseModel):
    status: Status
    role: Literal["admin", "user", "guest"]

a = Account(status="active", role="admin")
print(a.status)  # Status.active

6、环境变量配置(baseSettings)

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "MyApp"
    debug: bool = False
    database_url: str

    model_config = {
        "env_file": ".env",
        "env_prefix": "APP_",
    }

# .env 文件内容:
# APP_DATABASE_URL=postgresql://localhost/db
# APP_DEBUG=true

settings = Settings()
print(settings.database_url)

7、序列化与模型转换

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    password: str

u = User(id=1, name="Alice", password="secret")

# 导出时排除敏感字段
print(u.model_dump(exclude={"password"}))
# {'id': 1, 'name': 'Alice'}

# 从 ORM 对象创建(需开启 from_attributes)
class UserORM:
    def __init__(self, id, name, password):
        self.id = id
        self.name = name
        self.password = password

class UserSchema(BaseModel):
    id: int
    name: str
    model_config = {"from_attributes": True}

orm_user = UserORM(1, "Bob", "xxx")
print(UserSchema.model_validate(orm_user))

8、泛型模型

from pydantic import BaseModel
from typing import Generic, TypeVar

T = TypeVar("T")

class Response(BaseModel, Generic[T]):
    code: int
    message: str
    data: T

class UserInfo(BaseModel):
    id: int
    name: str

resp = Response[UserInfo](
    code=200,
    message="ok",
    data={"id": 1, "name": "Alice"}
)
print(resp.data.name)  # Alice