企业官网建设流程全解析

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习

本文介绍 openJiuwen agent-core 检索模块（openjiuwen.core.retrieval.embedding.dashscope_embedding）中的DashscopeEmbedding客户端。它基于阿里云 DashScope（dashscopeSDK）的多模态向量 API，支持文本、图片、视频等混合输入的统一向量化，是构建多模态知识库、图搜与智能体检索链路的关键组件。读完本文，你将掌握其全部构造参数、同步/异步嵌入 API、MultimodalDocument多模态文档构造方式、源码级批处理与并发原理，以及可直接运行的实战示例。

一、DashscopeEmbedding 是什么

DashscopeEmbedding是 openJiuwen agent-core 中对接阿里云百炼（DashScope）多模态向量 API 的嵌入客户端实现，源码位于 dashscope_embedding.py。

它的核心能力与定位包括：

• 多模态输入：单条或多条文档中既可以放纯文本，也可以放图片、视频、音频（经MultimodalDocument包装），统一调用百炼多模态向量 API 得到向量；
• 同步与异步双接口：每个嵌入方法都提供async版本与_sync后缀版本，适配 asyncio 与同步两种编程模型；
• 继承自通用 HTTP 嵌入客户端：类继承关系为Embedding（抽象基类）→ APIEmbedding（通用 HTTP 实现）→ DashscopeEmbedding，其中Embedding、EmbeddingConfig定义在 base_embedding.py，APIEmbedding定义在 api_embedding.py；
• 底层调用方式：同步请求走dashscope.MultiModalEmbedding.call，异步请求走dashscope.AioMultiModalEmbedding.call，两者通过dashscopeSDK 访问base_url配置的服务地址（默认形态为https://dashscope.aliyuncs.com/api/v1/）。

说明：本模块调用的具体模型与接口格式（如qwen3-vl-embedding等模型的输入输出规范）以阿里云百炼多模态向量 API 官方文档为准，仓库内的实现负责参数装配、批处理、重试与响应解析。

二、构造函数与参数详解

2.1 完整签名

DashscopeEmbedding( config: EmbeddingConfig, timeout: int = 60, max_retries: int = 3, extra_headers: Optional[dict] = None, max_batch_size: int = 8, max_concurrent: int = 50, dimension: Optional[int] = None, verify: bool | str | ssl.SSLContext = True, **kwargs, )

2.2 参数说明表

2.3 参数底层行为解析（源码佐证）

• dimension与 Matryoshka：构造时若传入整数dimension，实现会设置self.matryoshka_dimension = True，并将"dimension"写入self._request_params（dashscope_embedding.py）；若不传，_dimension保持None，在首次成功响应后由len(embeddings[0])推断并缓存（同文件_handle_dashscope_api_resp）。
• max_concurrent的双重作用：异步路径使用asyncio.Semaphore(max_concurrent)控制并发批处理；同步路径在APIEmbedding中创建ThreadPoolExecutor(max_workers=max_concurrent)（api_embedding.py）。
• verify与 SSL 环境变量：除显式参数外，基类还会读取环境变量EMBEDDING_SSL_VERIFY（置为false时关闭校验）与EMBEDDING_SSL_CERT（自定义 CA 证书路径）；自定义ssl.SSLContext会通过SSLContextAdapter挂载到 requests 会话、通过aiohttp.TCPConnector(ssl=...)用于异步连接（dashscope_embedding.py）。
• api_key可为空：EmbeddingConfig.api_key是可选字段（Optional[str]），基类仅在非空时才拼装Authorization头；测试test_init_without_api_key验证了该行为。
• 析构清理：__del__中会尝试关闭aio_connector与 requestsreq_session，避免连接泄漏。

三、EmbeddingConfig 配置

EmbeddingConfig是 pydantic 数据模型（base_embedding.py），三个字段均为字符串：

from openjiuwen.core.retrieval import EmbeddingConfig MULTIMODAL_EMBEDDING_CONFIG = EmbeddingConfig( api_key="<YOUR_DASHSCOPE_API_KEY>", base_url="https://dashscope.aliyuncs.com/api/v1/", model_name="qwen3-vl-embedding", )

在仓库的 examples/retrieval/configs.py 中，配置统一从.env加载：DASHSCOPE_API_KEY直接取自环境变量，而通用多模态配置使用MULTIMODAL_EMBEDDING_API_BASE、MULTIMODAL_EMBEDDING_API_KEY、MULTIMODAL_EMBEDDING_MODEL三个变量。完整的环境变量模板见 examples/retrieval/.env.example：

# examples/retrieval/showcase_dashscope_multimodal_embedding.py DASHSCOPE_API_KEY= # examples/retrieval/showcase_multimodal_embedding.py MULTIMODAL_EMBEDDING_API_BASE= MULTIMODAL_EMBEDDING_API_KEY= MULTIMODAL_EMBEDDING_MODEL=

安装dashscopeSDK 并完成密钥配置后，即可将EmbeddingConfig传入DashscopeEmbedding构造器。

四、核心 API 详解

4.1 property dimension

dimension -> int

返回嵌入向量的维度。若构造时指定了dimension，直接返回；否则首次调用会通过同步方式发送一条"test"查询来探测维度并缓存（api_embedding.py）。这保证了在任何上下文（包括异步代码）中读取该属性都是安全的。

4.2 单条查询嵌入

异步版本：

async def embed_query(text: str | MultimodalDocument, **kwargs: Any) -> List[float]

同步版本：

def embed_query_sync(text: str | MultimodalDocument, **kwargs: Any) -> List[float]

两者接收一条纯文本str或多模态文档，返回单个List[float]向量。实现上embed_query复用批量接口：embeddings = await self.embed_documents([text])，再取embeddings[0]。

4.3 批量嵌入

异步版本：

async def embed_documents( texts: List[str | MultimodalDocument], batch_size: Optional[int] = None, **kwargs: Any, ) -> List[List[float]]

同步版本：

def embed_documents_sync( texts: List[str | MultimodalDocument], batch_size: Optional[int] = None, **kwargs: Any, ) -> List[List[float]]

要点：

• 返回结果与输入texts顺序严格一致；
• batch_size为空时使用max_batch_size；实际批大小取batch_size与构造时max_batch_size的较小值，保证不超出 DashScope 接口批量上限（源码见 dashscope_embedding.py）；
• 传入元素为MultimodalDocument时，会先经validate_embed_docs校验（拒绝空列表、拒绝空字符串文档、校验callback_cls类型），再将其转换为doc.dashscope_input作为请求体中的input元素；
• kwargs中可携带callback_cls：必须是BaseCallback的子类（默认BaseCallback），每批完成后回调(start_idx, end_idx, batch)，可用于进度跟踪——仓库提供了现成的 TqdmCallback 用于展示进度条。

4.4 纯多模态嵌入接口

async def embed_multimodal(doc: MultimodalDocument, **kwargs) -> List[float] def embed_multimodal_sync(doc: MultimodalDocument, **kwargs) -> List[float]

这两个接口只接受MultimodalDocument，若传入其他类型（如str）会抛出RETRIEVAL_EMBEDDING_INPUT_INVALID错误，错误信息为input provided for multimodal embedding is not a MultimodalDocument（dashscope_embedding.py）。适合明确按“一条多模态文档 = 一个向量”的语义调用。

五、多模态输入：MultimodalDocument

多模态文档模型定义在 document.py，详细 API 文档见 document.md。

5.1 支持的模态与添加方式

add_field支持text、image、audio、video四种类型，数据来源可以是直接字符串（URL 或 base64 data URI）、本地文件路径Path，并可链式调用：

from openjiuwen.core.retrieval import MultimodalDocument from pathlib import Path # 文本 + 本地图片 doc = MultimodalDocument() doc.add_field("text", "A photograph of a person") doc.add_field("image", file_path=Path("reference.jpg")) # 链式写法 + 远程图片 URL doc2 = (MultimodalDocument() .add_field("text", "Picture of an octopus in ocean") .add_field("image", data="https://openjiuwen.com/img/jiuwen_logo.png")) # base64 音频数据 doc3 = MultimodalDocument() doc3.add_field("audio", data="data:audio/wav;base64,...")

add_field会校验模态类型、数据来源（file_path与data二选一）、文件存在性与 MIME 类型，并将本地文件转为 base64 数据；data_id用于多模态缓存，为空时自动生成 32 位十六进制 uuid。

5.2 DashScope 请求体格式

MultimodalDocument.dashscope_input属性将内部数据转换为 DashScope 多模态 API 的input元素格式（document.py）：

• text→{"text": <字符串>}；
• image→ 单张时为{"image": <url或base64>}，多张时为{"multi_images": [...]}；
• video→{"video": <url>}（DashScope 格式仅支持 URL 形式的视频，不支持 base64 视频，传入会报错）；
• 同一种模态不允许出现多个字段（text除外按单字段处理），否则抛出校验错误；
• 注意：audio不在dashscope_input支持范围内（dashscope_input只处理 text/image/video），如需音频嵌入应使用其他模型接口。

DashscopeEmbedding在批量请求前会把每个MultimodalDocument自动替换为doc.dashscope_input，因此调用方无需手动转换。

六、源码级原理：批处理、并发与重试

6.1 异步批处理与并发控制

embed_documents的异步实现（dashscope_embedding.py）：

1. 校验并转换输入；
2. 按bsz = min(batch_size, max_batch_size)切分成多个批；
3. 每个批在一个process_batch协程中执行，协程内部先获取async with self.limiter信号量（asyncio.Semaphore(max_concurrent)）再发请求，保证全局并发不超过max_concurrent；
4. 所有批通过asyncio.gather并行执行，每批完成后触发callback_obj(start_idx, end_idx, batch)；
5. 最后用chain.from_iterable将各批结果按序拼接返回。

6.2 同步批处理与线程池

embed_documents_sync使用ThreadPoolExecutor（线程名前缀openjiuwen_embed）并发提交各批请求，as_completed接收结果后按原索引回填，同样保证返回顺序与输入一致（dashscope_embedding.py）。

6.3 请求参数与重试

底层请求载荷为：

{ "model": self.model_name, "api_key": self.api_key, "base_address": self.api_url, "timeout": self.timeout, # "dimension": self._dimension, # 仅 Matryoshka 模式时存在 "input": payload_input, **kwargs, }

_get_embeddings/_get_embeddings_sync在range(self.max_retries)内循环调用 dashscope SDK；响应解析函数_handle_dashscope_api_resp会：

• 非 200 且已达最后一次重试时抛出RETRIEVAL_EMBEDDING_REQUEST_CALL_FAILED；
• 按index对响应中的embeddings排序，保证批次内顺序；
• 响应中embeddings为空或缺失时抛出RETRIEVAL_EMBEDDING_RESPONSE_INVALID；
• 首次成功响应时推断并缓存向量维度。

七、实战示例：多模态向量相似度验证

仓库 examples/retrieval/ 目录提供了三个可直接运行的示例，其中与本客户端直接相关的是showcase_dashscope_multimodal_embedding.py（阿里云 DashScope 多模态嵌入与向量相似度对比）。核心流程如下（完整代码见 showcase_dashscope_multimodal_embedding.py）：

import asyncio from pathlib import Path from configs import DASHSCOPE_API_KEY from utils.vector_similarities import cosine_similarity, euclidean_distance from openjiuwen.core.retrieval import DashscopeEmbedding, EmbeddingConfig, MultimodalDocument REFERENCE_TEXT = "A photograph of a person" DIFFERENT_TEXT = "Picture of an octopus in ocean" LOCAL_REF_IMAGE = Path("reference.jpg") DIFFERENT_IMAGE = "https://openjiuwen.com/img/jiuwen_logo.png" EMBEDDING_DIM = 256 # 置为 None 使用模型默认维度 MULTIMODAL_EMBEDDING_CONFIG = EmbeddingConfig( api_key=DASHSCOPE_API_KEY, base_url="https://dashscope.aliyuncs.com/api/v1/", model_name="qwen3-vl-embedding", ) async def main(): # 构造 3 个文档：文本相同但图片不同 / 图片相同但文本不同 docs = [MultimodalDocument() for _ in range(3)] docs[0].add_field("text", REFERENCE_TEXT).add_field("image", file_path=LOCAL_REF_IMAGE) docs[1].add_field("text", REFERENCE_TEXT).add_field("image", data=DIFFERENT_IMAGE) docs[2].add_field("text", DIFFERENT_TEXT).add_field("image", data=DIFFERENT_IMAGE) model = DashscopeEmbedding(MULTIMODAL_EMBEDDING_CONFIG, dimension=EMBEDDING_DIM, timeout=30) emb1, emb2, emb3 = await model.embed_documents(docs) print("Embedding dimensions:", len(emb1)) # 不同图片的相似度应明显更低 sim_1_2 = cosine_similarity(emb1, emb2) # 同文本、不同图片 sim_2_3 = cosine_similarity(emb2, emb3) # 同图片、不同文本 print("same text/diff image similarity: %.4f" % sim_1_2) print("same image/diff text similarity: %.4f" % sim_2_3) if __name__ == "__main__": asyncio.run(main())

该示例的语义设计很有参考价值：

• doc1与doc2文本完全相同、图片不同，用于验证图片对向量的显著影响；
• doc2与doc3图片相同、文本不同，用于对比文本对向量的影响权重；
• 运行后可用余弦相似度与欧氏距离两套指标判断：不同图片的相似度通常应低于 0.9，且“同图不同文”的相似度应高于“异图异文”。

同目录下的 showcase_multimodal_embedding.py 演示了VLLMEmbedding的embed_multimodal用法（含同图不同格式、不同图片、不同文本的对比），showcase_text_embedding.py 则演示纯文本批量嵌入与跨语言语义相似度验证，两者可与 DashScope 示例相互对照。

八、测试验证与异常处理

仓库为DashscopeEmbedding提供了完整的单元测试，见 tests/unit_tests/core/retrieval/embedding/test_dashscope_embedding.py，覆盖：

使用中的常见异常与应对建议：

• 空输入：embed_documents([])或包含空字符串的列表会抛RETRIEVAL_EMBEDDING_INPUT_INVALID（"Empty texts list provided" / "chunks are empty while embedding"），调用前应过滤空文档；
• 非法回调：callback_cls不是BaseCallback子类时抛RETRIEVAL_EMBEDDING_CALLBACK_INVALID；
• 请求失败：超过max_retries后抛RETRIEVAL_EMBEDDING_REQUEST_CALL_FAILED，可按业务调整max_retries与timeout；
• 响应异常：接口返回中缺少或为空embeddings字段时抛RETRIEVAL_EMBEDDING_RESPONSE_INVALID，通常是模型名或输入格式不匹配所致，应核对百炼多模态向量 API 文档中的模型可用性与输入规范。

九、总结与使用建议

DashscopeEmbedding为 openJiuwen agent-core 的检索链路提供了开箱即用的阿里云多模态向量化能力，其设计要点可总结为：

1. 统一入口：文本与多模态文档共用embed_documents/embed_query接口，返回顺序严格一致，便于直接对接向量库索引；
2. 可控的规模化：max_batch_size保证单请求不超限，max_concurrent限制总体并发，配合TqdmCallback可在索引海量文档时清晰观察进度；
3. 同步异步双模：同步路径基于线程池，异步路径基于aiohttp+ 信号量，同一套参数即可适配不同运行环境；
4. 灵活的向量维度：支持 Matryoshka 类模型指定维度，也支持自动探测，兼顾存储成本与召回效果。

在实际落地时，建议将密钥放入.env管理、在构造器显式设置timeout与max_retries，并优先使用MultimodalDocument.add_field的链式写法组织图片与文本；若需深度定制 HTTPS 校验（如内网自签证书环境），可通过verify参数或EMBEDDING_SSL_CERT环境变量配置。

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习