企业官网建设流程全解析

• 人工智能
• 大模型
• AI Agent
• RAG
• 后端
• 前端

导读

本文围绕 TaskingAI 后端代码模板中面向异步 Python SDK的文本嵌入（Text Embedding）示例（backend/resources/code_templates/model/2_async_python/1_inference/2_text_embedding.md），系统讲解如何使用taskingai的异步接口taskingai.inference.a_text_embedding完成单条与批量文本向量化。文章同时结合仓库中后端路由、推理服务与多供应商适配源码，说明请求参数、响应结构、批处理与向量归一化等底层机制，帮助读者在 RAG 检索、语义匹配等场景中正确、高效地使用 TaskingAI 的嵌入能力。

使用前提

在调用嵌入接口之前，需要先确认两件事：

1. 已安装taskingaiPython SDK，并在代码中导入。
2. 拥有一个可用的文本嵌入模型：在 TaskingAI 控制台完成模型配置后，会得到一个 8 位长度的model_id（如abcdefgh）。该 ID 由后端请求模型在 backend/app/schemas/model/text_embedding.py 中约束为min_length=8, max_length=8。

注意：本文示例中的$$MODEL_ID$$为占位符，请替换为你实际创建并配置好的嵌入模型 ID。

单条文本嵌入

异步 Python SDK 的单条调用示例如下：

import taskingai embedding_result = await taskingai.inference.a_text_embedding( model_id="$$MODEL_ID$$", input="Machine learning is a subfield of artificial intelligence (AI) that involves the development of algorithms that allow computers to learn from and make decisions or predictions based on data." )

与同步接口taskingai.inference.text_embedding(...)（见 backend/resources/code_templates/model/1_python/1_inference/2_text_embedding.md）相比，异步版本：

• 以await挂起协程，不阻塞事件循环，适合高并发应用；
• 在async上下文中执行，常用于 FastAPI 路由、异步数据处理管道等场景。

调用后返回的embedding_result中包含data（嵌入向量列表）与usage（Token 用量）两部分。

批量文本嵌入

将input参数改为字符串列表，即可一次请求为多条文本生成向量：

embedding_list = await taskingai.inference.a_text_embedding( model_id="$$MODEL_ID$$", input=[ "Machine learning is a subfield of artificial intelligence (AI) that involves the development of algorithms that allow computers to learn from and make decisions or predictions based on data.", "Michael Jordan, often referred to by his initials MJ, is considered one of the greatest players in the history of the National Basketball Association (NBA)." ] )

input参数在请求模型中定义为Union[str, List[str]]（见 backend/app/schemas/model/text_embedding.py），单条字符串与字符串列表均可直接传入。

核心参数说明

除model_id与input外，文本嵌入请求还支持input_type：

input_type的取值定义在 backend/app/models/inference/text_embedding.py 中，枚举为document与query：

• document：待入库的知识文档片段，适用于索引/写入侧；
• query：检索时的查询语句，适用于检索侧。

实际是否需要传递该字段取决于底层模型供应商是否支持区分查询与文档的嵌入策略；TaskingAI 会将其透传给适配的供应商实现（见 inference/app/routes/text_embedding/schema.py）。

请求处理链路与响应结构

后端路由：模型校验与 fallback

TaskingAI 后端将/v1/inference/text_embedding请求封装在 backend/app/routes/inference/text_embedding.py 中。核心流程为：

1. 根据model_id加载模型实体；
2. 组装[主模型] + [fallback 模型列表]的尝试顺序；
3. 依次调用 backend/app/services/inference/text_embedding.py 中的text_embedding(...)服务函数；
4. 主模型失败时自动切换到 fallback 模型，全部失败则返回PROVIDER_ERROR。

服务函数将请求转发到独立的 Inference 服务：{TASKINGAI_INFERENCE_URL}/v1/text_embedding，载荷中包含model_schema_id、provider_model_id、encrypted_credentials、properties、input与可选的input_type。

响应结构

TextEmbeddingResponse（见 backend/app/schemas/model/text_embedding.py）结构如下：

{ "status": "success", "data": [ { "index": 0, "embedding": [0.1, 0.2, 0.3, 0.4] } ], "usage": { "input_tokens": 12 } }

• data：嵌入向量列表，index与输入顺序一一对应，embedding为浮点向量；
• usage.input_tokens：输入文本的 Token 用量，用于成本统计。

在 Inference 服务内部，usage通过string_tokens()（见 inference/app/models/tokenizer.py）基于cl100k_base编码器逐条统计 Token 数。

批量请求的底层处理：分批与并行

Inference 服务在 inference/app/routes/text_embedding/route.py 中对批量输入做了两层处理：

1. 分批：按模型属性max_batch_size（默认 512）将输入列表切片；该属性定义于 inference/app/models/text_embedding/model_config.py，即“供应商单次 API 调用能处理的最大文本块数量”；
2. 并行：以 20 个批次为上限，通过asyncio.gather并发执行嵌入任务，再按原始顺序合并结果。

这意味着：即使一次传入上千条文本，TaskingAI 也会自动拆分并发请求，同时保证返回结果的index顺序与输入一致。

向量归一化

每个批次返回后，Inference 服务还会将嵌入向量强制归一化为单位向量（见 inference/app/routes/text_embedding/route.py）：以numpy.linalg.norm计算模长后逐向量除模。归一化后，余弦相似度与内积等价，这也是检索场景中相似度计算可以简化的原因。

供应商适配示例：以 OpenAI 为例

TaskingAI 的 Inference 服务为每家供应商实现embed_text()抽象方法（基类见 inference/app/models/text_embedding/model.py）。以 inference/providers/openai/text_embedding.py 为例：

• 请求 OpenAI/v1/embeddings，携带Authorization: Bearer <OPENAI_API_KEY>；
• 支持text-embedding-3-small/text-embedding-3-large的维度变体（如text-embedding-3-small-512会通过dimensions参数请求 512 维向量）；
• 支持通过proxy参数覆盖上游 URL，以及custom_headers注入自定义请求头。

因此，同样的a_text_embedding调用，只要model_id指向不同供应商的嵌入模型，即可无缝切换到底层供应商实现。

多供应商与测试覆盖

仓库在 inference/providers 下为 20+ 家供应商提供了文本嵌入适配实现，包括 OpenAI、Anthropic 系列之外的 Azure OpenAI、AWS Bedrock、Google Gemini、Cohere、Jina、Mistral AI、Ollama、Zhipu、Tongyi 等，均可通过统一的a_text_embedding接口调用。

后端 API 测试 backend/tests/api_services/inference/text_embedding.py 与服务测试 backend/tests/services_tests/inference/test_text_embedding.py 覆盖了：

• 单条与批量输入的成功路径（断言状态码 200 且embedding中均为浮点数）；
• 长文本列表（每条 600 字符）与空列表[]的边界情况；
• 非法输入（如input=100、列表中混入整数100）返回 422 与REQUEST_VALIDATION_ERROR。

这些测试同时验证了主模型与 fallback 模型（如fallbacks_text_embedding_model_id）的容错路径。

延伸：OpenAI 兼容端点

除 TaskingAI 原生接口外，仓库还提供 OpenAI 兼容的/v1/embeddings端点（实现见 backend/app/routes/openai/text_embedding.py，模板见 backend/resources/code_templates/model/4_openai_compatible_api/1_stateless_invocation/2_text_embedding.md），可直接使用openaiPython 客户端调用：

from openai import OpenAI client = OpenAI( api_key="$$TASKINGAI_API_KEY$$", base_url="$$SERVICE_HOST$$/v1", ) response = client.embeddings.create( model="$$MODEL_ID$$", input=["Hello!"], )

该端点内部会将 OpenAI 请求格式适配为 TaskingAI 请求格式（见 backend/app/routes/openai/utils/adapt_text_embedding.py），并把usage.prompt_tokens映射为input_tokens。

总结

本文基于 TaskingAI 异步 Python 代码模板，完整演示了taskingai.inference.a_text_embedding的单条与批量调用方式，并结合后端与 Inference 服务源码说明了参数语义、fallback 容错、批量分批并发、向量归一化与 Token 统计等实现细节。在实际项目中，你可以将返回的嵌入向量用于向量检索库的索引与查询，或作为语义相似度计算的输入；如需进一步了解检索相关能力，可继续阅读 backend/resources/code_templates/collection 下的集合与记录代码模板。

• 人工智能
• 大模型
• AI Agent
• RAG
• 后端
• 前端