让我们更详细地看看如何使用 DocArray 表示您的数据：

from docarray import BaseDoc
from docarray.typing import TorchTensor, ImageUrl
from typing import Optional
import torch


# 定义您的数据模型
class MyDocument(BaseDoc):
    description: str
    image_url: ImageUrl  # 也可以是 VideoUrl、AudioUrl 等
    image_tensor: Optional[
        TorchTensor[1704, 2272, 3]
    ] = None  # 也可以是 NdArray 或 TensorflowTensor
    embedding: Optional[TorchTensor] = None

因此，您不仅可以定义数据的类型，还可以 指定张量的形状！

# 创建一个文档
doc = MyDocument(
    description="这是一张山的照片",
    image_url="https://upload.wikimedia.org/wikipedia/commons/2/2f/Alpamayo.jpg",
)

# 从 URL 加载图像张量
doc.image_tensor = doc.image_url.load()


# 使用您选择的任何模型计算嵌入
def clip_image_encoder(image_tensor: TorchTensor) -> TorchTensor:  # 虚拟函数
    return torch.rand(512)


doc.embedding = clip_image_encoder(doc.image_tensor)

print(doc.embedding.shape)  # torch.Size([512])

组合嵌套的 Document

当然，你可以将 Document 组合成嵌套结构：

from docarray import BaseDoc
from docarray.documents import ImageDoc, TextDoc
import numpy as np


class MultiModalDocument(BaseDoc):
    image_doc: ImageDoc
    text_doc: TextDoc


doc = MultiModalDocument(
    image_doc=ImageDoc(tensor=np.zeros((3, 224, 224))), text_doc=TextDoc(text='hi!')
)

在实际应用中，尤其是机器学习领域，很少会单独处理一个数据点。因此，你可以轻松地收集多个 Document：

收集多个 Document

在构建或与机器学习系统交互时，通常需要一次处理多个 Document（即多个数据点）。

DocArray 提供了两种数据结构来实现这一点：

• DocVec：一组 Document 的向量。所有 Document 中的张量会被堆叠成一个单一的张量。非常适合批量处理和在机器学习模型中使用。
• DocList：一组 Document 的列表。所有 Document 中的张量保持原样。非常适合流式传输、重新排序和数据洗牌。

我们先来看看 DocVec：

from docarray import DocVec, BaseDoc
from docarray.typing import AnyTensor, ImageUrl
import numpy as np


class Image(BaseDoc):
    url: ImageUrl
    tensor: AnyTensor  # 这允许使用 PyTorch、NumPy 和 TensorFlow 张量


vec = DocVec[Image](  # DocVec 根据你的自定义 Schema 进行参数化！
    [
        Image(
            url="https://upload.wikimedia.org/wikipedia/commons/2/2f/Alpamayo.jpg",
            tensor=np.zeros((3, 224, 224)),
        )
        for _ in range(100)
    ]
)

在上面的代码片段中，DocVec 是根据你想要使用的 Document 类型进行参数化的：DocVec[Image]。

这看起来可能有些奇怪，但我们相信你会很快习惯！此外，它还允许我们做一些很酷的事情，比如批量访问你在 Document 中定义的字段：

tensor = vec.tensor  # 获取 DocVec 中的所有张量
print(tensor.shape)  # 它们被堆叠成了一个单一的张量！
print(vec.url)  # 你也可以批量访问其他字段

第二种数据结构 DocList 的工作方式类似：

from docarray import DocList

dl = DocList[Image](  # DocList 根据你的自定义 Schema 进行参数化！
    [
        Image(
            url="https://upload.wikimedia.org/wikipedia/commons/2/2f/Alpamayo.jpg",
            tensor=np.zeros((3, 224, 224)),
        )
        for _ in range(100)
    ]
)

你仍然可以批量访问 Document 中的字段：

tensors = dl.tensor  # 获取 DocList 中的所有张量
print(type(tensors))  # 以张量列表的形式
print(dl.url)  # 你也可以批量访问其他字段

此外，你还可以向 DocList 中插入、删除或追加 Document：

# 追加
dl.append(
    Image(
        url="https://upload.wikimedia.org/wikipedia/commons/2/2f/Alpamayo.jpg",
        tensor=np.zeros((3, 224, 224)),
    )
)
# 删除
del dl[0]
# 插入
dl.insert(
    0,
    Image(
        url="https://upload.wikimedia.org/wikipedia/commons/2/2f/Alpamayo.jpg",
        tensor=np.zeros((3, 224, 224)),
    ),
)

你还可以无缝地在 DocVec 和 DocList 之间切换：

vec_2 = dl.to_doc_vec()
assert isinstance(vec_2, DocVec)

dl_2 = vec_2.to_doc_list()
assert isinstance(dl_2, DocList)

如果您正在使用 DocArray 0.30.0 或更低版本，您应该熟悉其 数据类 API。

DocArray ≥0.30 是这一理念的进一步发展。 每个文档都是通过类似数据类的接口创建的， 这得益于 Pydantic。

这带来了以下优势：

• 灵活性： 无需遵循固定的字段集——您的数据定义了模式。
• 多模态性： 文档本质上只是字典。这使得它们可以轻松地从任何语言创建并发送，而不仅仅是 Python。

您可能也熟悉我们用于向量数据库集成的旧版文档存储。 现在它们被称为 文档索引，并提供了以下改进（有关新 API，请参阅 此处）：

• 混合搜索： 您现在可以将向量搜索与文本搜索结合，甚至可以根据任意字段进行过滤。
• 生产就绪： 新的文档索引是对各种向量数据库库的更轻量封装，使其更加健壮且易于维护。
• 更高的灵活性： 我们致力于支持您可以通过数据库官方客户端执行的任何配置或设置。

目前，文档索引支持 Weaviate、Qdrant、ElasticSearch、Redis、Mongo Atlas、精确最近邻搜索以及 HNSWLib，未来还将支持更多。

如果您来自 Pydantic 社区，可以将 DocArray 文档视为功能增强的 Pydantic 模型，而 DocArray 则是围绕这些模型的一系列扩展工具。

具体来说，我们的目标是 让 Pydantic 更适合机器学习领域——不是取代它，而是在此基础上构建！

这意味着您将获得以下好处：

• 面向机器学习的类型： Tensor、TorchTensor、Embedding 等，包括 张量形状验证。
• 与 FastAPI volle 兼容。
• DocList 和 DocVec 将模型的概念推广到模型的 序列 或 批次。非常适合 用于机器学习模型 和其他批量处理任务。
• 活体类型： ImageUrl 可以 .load() URL 并转换为图像张量，TextUrl 可以加载并分词文本文档等。
• 云就绪：支持序列化为 Protobuf，适用于微服务和 gRPC。
• 预建的多模态文档，适用于不同数据模态：图像、文本、3D网格、视频、音频等。请注意，所有这些都符合 Pydantic 模型的标准！
• 文档存储 和 文档索引 让您能够存储数据，并通过 向量搜索 进行检索。

这里最明显的优势是 对以机器学习为中心的数据提供一流的支持，例如 {Torch, TF, ...}Tensor、Embedding 等。

这还包括一些实用的功能，比如验证张量的形状：

from docarray import BaseDoc
from docarray.typing import TorchTensor
import torch


class MyDoc(BaseDoc):
    tensor: TorchTensor[3, 224, 224]


doc = MyDoc(tensor=torch.zeros(3, 224, 224))  # 成功
doc = MyDoc(tensor=torch.zeros(224, 224, 3))  # 通过重塑成功

try:
    doc = MyDoc(tensor=torch.zeros(224))  # 验证失败
except Exception as e:
    print(e)
    # tensor
    # 无法将形状为 (224,) 的张量重塑为 (3, 224, 224)（类型=value_error）


class Image(BaseDoc):
    tensor: TorchTensor[3, 'x', 'x']


Image(tensor=torch.zeros(3, 224, 224))  # 成功

try:
    Image(
        tensor=torch.zeros(3, 64, 128)
    )  # 验证失败，因为第二维度与第三维度不匹配
except Exception as e:
    print()


try:
    Image(
        tensor=torch.zeros(4, 224, 224)
    )  # 验证失败，因为第一维度不符合要求
except Exception as e:
    print(e)
    # 张量形状不匹配。预期 (3, 'x', 'x')，实际得到 (4, 224, 224)（类型=value_error）

try:
    Image(
        tensor=torch.zeros(3, 64)
    )  # 验证失败，因为维度不足
except Exception as e:
    print(e)
    # 张量形状不匹配。预期 (3, 'x', 'x')，实际得到 (3, 64)（类型=value_error）

如果你来自 PyTorch 生态，可以将 DocArray 看作一种 在数据流经模型时组织数据 的方式。

它为你提供了多项优势：

• 在类型提示中表达 张量的形状
• 将属于同一对象的张量分组，例如一段音频和一张图像
• 直接部署上线，通过复用你的数据模型作为 FastAPI 或 Jina API 的 Schema
• 使用 Protobuf 和 gRPC 在 微服务之间 连接模型组件

DocArray 可以直接用于机器学习模型中，以处理和表示多模态数据。这使得你能够在 nn.Module 的深层逻辑中利用 DocArray 的抽象来分析数据，并提供与 FastAPI 兼容的 Schema，从而简化从模型训练到模型推理部署的过渡。

为了更好地理解这一点，我们先来看一个原生 PyTorch 实现的三模态机器学习模型：

import torch
from torch import nn


def encoder(x):
    return torch.rand(512)


class MyMultiModalModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.audio_encoder = encoder()
        self.image_encoder = encoder()
        self.text_encoder = encoder()

    def forward(self, text_1, text_2, image_1, image_2, audio_1, audio_2):
        embedding_text_1 = self.text_encoder(text_1)
        embedding_text_2 = self.text_encoder(text_2)

        embedding_image_1 = self.image_encoder(image_1)
        embedding_image_2 = self.image_encoder(image_2)

        embedding_audio_1 = self.image_encoder(audio_1)
        embedding_audio_2 = self.image_encoder(audio_2)

        return (
            embedding_text_1,
            embedding_text_2,
            embedding_image_1,
            embedding_image_2,
            embedding_audio_1,
            embedding_audio_2,
        )

坦白说，这段代码可读性并不高。更糟糕的是，如果需要再增加一种模态，你就不得不修改整个代码库：既要更改 forward() 方法的返回值类型，还要对后续所有依赖这部分输出的代码进行大量调整。

接下来，让我们看看使用 DocArray 后同样的代码会是什么样子：

from docarray import DocList, BaseDoc
from docarray.documents import ImageDoc, TextDoc, AudioDoc
from docarray.typing import TorchTensor
from torch import nn
import torch


def encoder(x):
    return torch.rand(512)


class Podcast(BaseDoc):
    text: TextDoc
    image: ImageDoc
    audio: AudioDoc


class PairPodcast(BaseDoc):
    left: Podcast
    right: Podcast


class MyPodcastModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.audio_encoder = encoder()
        self.image_encoder = encoder()
        self.text_encoder = encoder()

    def forward_podcast(self, docs: DocList[Podcast]) -> DocList[Podcast]:
        docs.audio.embedding = self.audio_encoder(docs.audio.tensor)
        docs.text.embedding = self.text_encoder(docs.text.tensor)
        docs.image.embedding = self.image_encoder(docs.image.tensor)

        return docs

    def forward(self, docs: DocList[PairPodcast]) -> DocList[PairPodcast]:
        docs.left = self.forward_podcast(docs.left)
        docs.right = self.forward_podcast(docs.right)

        return docs

是不是清晰多了？代码的可读性和可维护性瞬间提升。而且只需稍加改动，你就能将 PyTorch 模型转换为 FastAPI 应用，并复用你的 Document Schema 定义（详见下方“来自 FastAPI”部分）。这一切都通过 Python 式的类型提示来实现。

与 PyTorch 方案 类似，你也可以在 TensorFlow 中使用 DocArray 来处理和表示多模态数据。

首先，要在 TensorFlow 中使用 DocArray，你需要按照以下步骤安装：

pip install tensorflow==2.12.0
pip install protobuf==3.19.0

与在 PyTorch 中使用 DocArray 相比，在 TensorFlow 中使用时有一个主要区别：DocArray 的 TorchTensor 是 torch.Tensor 的子类，而 TensorFlowTensor 则不是。由于 tf.Tensor 的一些技术限制，DocArray 的 TensorFlowTensor 并非 tf.Tensor 的子类，而是将其存储在 .tensor 属性中。

这会对你产生什么影响呢？每当你需要访问张量数据——比如对其进行操作或传递给你的机器学习模型时——就不能直接传递 TensorFlowTensor 实例，而必须通过其 .tensor 属性来获取张量本身。

示例如下：

from typing import Optional

from docarray import DocList, BaseDoc

import tensorflow as tf


class Podcast(BaseDoc):
    audio_tensor: Optional[AudioTensorFlowTensor] = None
    embedding: Optional[AudioTensorFlowTensor] = None


class MyPodcastModel(tf.keras.Model):
    def __init__(self):
        super().__init__()
        self.audio_encoder = AudioEncoder()

    def call(self, inputs: DocList[Podcast]) -> DocList[Podcast]:
        inputs.audio_tensor.embedding = self.audio_encoder(
            inputs.audio_tensor.tensor
        )  # 访问 audio_tensor 的 .tensor 属性
        return inputs

文档是 Pydantic 模型（略有不同），因此它们与 FastAPI 完全兼容！

但为什么你应该使用它们，而不是你已经熟悉并喜爱的 Pydantic 模型呢？ 这是个好问题！

• 因为它们具备以机器学习为中心的功能、类型和验证，详见此处
• 因为 DocArray 可以充当向量数据库的 ORM，类似于 SQLModel 对 SQL 数据库的作用

为了进一步说明，我们来展示文档如何轻松地融入你的 FastAPI 应用：

import numpy as np
from fastapi import FastAPI
from docarray.base_doc import DocArrayResponse
from docarray import BaseDoc
from docarray.documents import ImageDoc
from docarray.typing import NdArray, ImageTensor


class InputDoc(BaseDoc):
    img: ImageDoc
    text: str


class OutputDoc(BaseDoc):
    embedding_clip: NdArray
    embedding_bert: NdArray


app = FastAPI()


def model_img(img: ImageTensor) -> NdArray:
    return np.zeros((100, 1))


def model_text(text: str) -> NdArray:
    return np.zeros((100, 1))


@app.post("/embed/", response_model=OutputDoc, response_class=DocArrayResponse)
async def create_item(doc: InputDoc) -> OutputDoc:
    doc = OutputDoc(
        embedding_clip=model_img(doc.img.tensor), embedding_bert=model_text(doc.text)
    )
    return doc


input_doc = InputDoc(text='', img=ImageDoc(tensor=np.random.random((3, 224, 224))))

async with AsyncClient(app=app, base_url="http://test") as ac:
    response = await ac.post("/embed/", data=input_doc.json())

就像普通的 Pydantic 模型一样！

Jina 已经采用 DocArray 作为其用于表示和序列化文档的库。

Jina 允许你部署和扩展使用 DocArray 构建的模型和服务，从而充分利用 DocArray 的序列化能力。

import numpy as np
from jina import Deployment, Executor, requests
from docarray import BaseDoc, DocList
from docarray.documents import ImageDoc
from docarray.typing import NdArray, ImageTensor


class InputDoc(BaseDoc):
    img: ImageDoc
    text: str


class OutputDoc(BaseDoc):
    embedding_clip: NdArray
    embedding_bert: NdArray


def model_img(img: ImageTensor) -> NdArray:
    return np.zeros((100, 1))


def model_text(text: str) -> NdArray:
    return np.zeros((100, 1))


class MyEmbeddingExecutor(Executor):
    @requests(on='/embed')
    def encode(self, docs: DocList[InputDoc], **kwargs) -> DocList[OutputDoc]:
        ret = DocList[OutputDoc]()
        for doc in docs:
            output = OutputDoc(
                embedding_clip=model_img(doc.img.tensor),
                embedding_bert=model_text(doc.text),
            )
            ret.append(output)
        return ret


with Deployment(
    protocols=['grpc', 'http'], ports=[12345, 12346], uses=MyEmbeddingExecutor
) as dep:
    resp = dep.post(
        on='/embed',
        inputs=DocList[InputDoc](
            [InputDoc(text='', img=ImageDoc(tensor=np.random.random((3, 224, 224)))]
        ),
        return_type=DocList[OutputDoc],
    )
    print(resp)

如果你将 DocArray 视为通用的向量数据库客户端，那么你可以把它看作一种新型的向量数据库 ORM。 DocArray 的作用是将多模态、嵌套且领域特定的数据映射到向量数据库中，在那里存储并使其可搜索：

from docarray import DocList, BaseDoc
from docarray.index import HnswDocumentIndex
import numpy as np

from docarray.typing import ImageUrl, ImageTensor, NdArray


class ImageDoc(BaseDoc):
    url: ImageUrl
    tensor: ImageTensor
    embedding: NdArray[128]


# 创建一些数据
dl = DocList[ImageDoc](
    [
        ImageDoc(
            url="https://upload.wikimedia.org/wikipedia/commons/2/2f/Alpamayo.jpg",
            tensor=np.zeros((3, 224, 224)),
            embedding=np.random.random((128,)),
        )
        for _ in range(100)
    ]
)

# 创建一个文档索引
index = HnswDocumentIndex[ImageDoc](work_dir='/tmp/test_index2')


# 索引你的数据
index.index(dl)

# 查找相似的文档
query = dl[0]
results, scores = index.find(query, limit=10, search_field='embedding')

目前，DocArray 支持以下向量数据库：

• Weaviate
• Qdrant
• Elasticsearch v8 和 v7
• Redis
• Milvus
• ExactNNMemorySearch 作为一种本地替代方案，提供精确的 kNN 搜索。
• HNSWlib 作为一种本地优先的近似最近邻搜索替代方案
• Mongo Atlas

目前正在开发对 OpenSearch 的集成。

当然，这仅仅是 DocArray 能够做到的事情之一，所以我们鼓励你查看本 README 的其余部分！

借助 DocArray，你可以通过 Langchain 将外部数据连接到 LLM。DocArray 让你能够自由定义灵活的文档模式，并从不同的后端中选择文档存储方式。 创建文档索引后，你可以使用 DocArrayRetriever 将其连接到你的 Langchain 应用程序。

通过以下命令安装 Langchain：

pip install langchain

1. 定义模式并创建文档：

from docarray import BaseDoc, DocList
from docarray.typing import NdArray
from langchain.embeddings.openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()


# 定义文档模式
class MovieDoc(BaseDoc):
    title: str
    description: str
    year: int
    embedding: NdArray[1536]


movies = [
    {"title": "#1 title", "description": "#1 description", "year": 1999},
    {"title": "#2 title", "description": "#2 description", "year": 2001},
]

# 嵌入“description”并创建文档
docs = DocList[MovieDoc](
    MovieDoc(embedding=embeddings.embed_query(movie["description"]), **movie)
    for movie in movies
)

2. 使用任何支持的后端初始化文档索引：

from docarray.index import (
    InMemoryExactNNIndex,
    HnswDocumentIndex,
    WeaviateDocumentIndex,
    QdrantDocumentIndex,
    ElasticDocIndex,
    RedisDocumentIndex,
    MongoDBAtlasDocumentIndex,
)

# 选择合适的后端并用数据初始化
db = InMemoryExactNNIndex[MovieDoc](docs)

3. 最后，初始化检索器并将其集成到你的链中！

from langchain.chat_models import ChatOpenAI
from langchain.chains import ConversationalRetrievalChain
from langchain.retrievers import DocArrayRetriever

# 创建一个检索器
retriever = DocArrayRetriever(
    index=db,
    embeddings=embeddings,
    search_field="embedding",
    content_field="description",
)

# 在你的链中使用该检索器
model = ChatOpenAI()
qa = ConversationalRetrievalChain.from_llm(model, retriever=retriever)

或者，你也可以使用内置的向量存储。Langchain 支持两种向量存储：DocArrayInMemorySearch 和 DocArrayHnswSearch。 这两种存储都易于使用，最适合中小型数据集。