🆕 新功能
-
优化动态批处理处理方式 (#6066)
为了改善系统的性能,我们优化了动态批处理的行为,确保不会把超过
preferred_batch_size
的文档发送给 Executor。这样一来,客户端也将能够更快地收到他们的请求的响应,不需要等待其他请求的处理完成。这样有助于降低系统的平均延迟,提高用户体验。 -
向 GatewayStreamer 方法添加
return_type
参数 (#6027)我们在 GatewayStreamer 方法中添加了一个
return_type
参数,用来指定返回结果的类型。在默认情况下,GatewayStreamer 会自动获取 Executor 的输入和输出模式,并动态重建 DocArray 模型,并将输出结果用于在 Gateway 级别上进行类型转换。虽然转换后的结果与 Executor 级别上定义的原始模式几乎相同,但可能会在某些类型检查上遇到错误。为了避免这种情况,现在可以在 GatewayStreamer 中指定
return_type
参数,确切地指定输出结果的类型。 -
Executor 可以处理单个文档 (#5991)
现在我们提供了用于处理单个文档的 Endpoint,也就是说 Executors 现在也可以处理单个 BaseDoc 了,不再仅限于 DocList。
这样一来,我们就更容易地理解和管理 Executor 的逻辑和功能了,不需要考虑批量处理多个文档的复杂性。特别是有些模型在处理单个文档时表现更好,这样就提供了更大的灵活性,开发者能够根据具体需求定制 Executor 的行为。
要实现这一点,
requests
装饰的方法需要接受一个doc
参数,并且使用输入和输出的类型注解:from jina import Executor, requests from docarray import BaseDoc class MyInputDocument(BaseDoc): num: int class MyOutputDocument(BaseDoc): label: str class MyExecutor(Executor): @requests(on='/hello') async def task(self, doc: MyInputDocument, **kwargs) -> MyOutputDocument: return MyOutputDocument(label='even' if doc.num % 2 == 0 else 'odd')
-
参数可以描述为 Pydantic 模型 (#6001)
Executor 参数可以使用 Pydantic 模型进行描述,而不是简单的 Python 字典,在定义时需要将模型作为类型注释。
这样有几个主要好处:
-
参数验证和自定义默认值:使用 Pydantic 模型作为参数的定义,可以提供参数的验证和默认值功能。这样就可以轻松地定义参数的默认值,减少了处理缺失或无效参数的繁琐工作。
-
更清晰的文档和类型提示:Pydantic 提供了强大的类型推断和自动文档生成功能,使得开发者可以快速了解参数的结构、类型以及可能的取值范围,提高代码的可读性和可维护性。
-
与 HTTP 协议的集成更加友好:在使用 HTTP 协议时,使用 Pydantic 模型作为参数的定义可以生成更具描述性的 OpenAPI 文档。开发者可以更轻松地了解 Executor 提供的接口规范。
-
-
更丰富的 OpenAPI (#5992)
当使用 HTTP 协议为 Executor 提供服务时,Executor 的 OpenAPI 描述变得更加丰富和详细了。我们给 Executor 的接口文档提供了更多详细的信息,包括描述、示例和其他相关字段。这些信息可以帮助开发者更准确地了解如何与 Executor 进行交互,并根据需要构建相应的请求。
-
在单一 Executor Flows 中支持流式传输 (#5988)
现在,流式传输端点还支持 Flow 编排层,并且不再强制使用 Deployment。Flow 编排层可以接受 gRPC 和 HTTP 协议下的流式传输端点。
with Flow(protocol=protocol, port=port, cors=True).add( uses=StreamingExecutor, ): client = Client(port=port, protocol=protocol, asyncio=True) i = 10 async for doc in client.stream_doc( on='/hello', inputs=MyDocument(text='hello world', number=i), return_type=MyDocument, ): print(doc)
-
使用 gRPC 协议的流式端点 (#5921)
之前,我们为 HTTP 协议添加了 SSE 支持,支持了流式传输文档,现在 gRPC 协议也支持了相同的功能。Jina 服务端可以使用 gRPC 协议依次向客户端流式传输单个文档。这个功能在内部依赖于流式传输的 gRPC 端点。
一个该功能的典型应用场景是在大型语言模型,流式地传输 token。可以查看我们关于如何流式传输 LLM Token 的文档。
from jina import Executor, requests, Deployment from docarray import BaseDoc # 首先定义模式 class MyDocument(BaseDoc): text: str # 然后定义 Executor class MyExecutor(Executor): @requests(on='/hello') async def task(self, doc: MyDocument, **kwargs) -> MyDocument: for i in range(100): yield MyDocument(text=f'hello world {i}') with Deployment( uses=MyExecutor, port=12345, protocol='grpc', # 或 'http' ) as dep: dep.block()
在客户端中,可以使用新的
stream_doc()
方法逐个接收文档:from jina import Client, Document client = Client(port=12345, protocol='grpc', asyncio=True) async for doc in client.stream_doc( on='/hello', inputs=MyDocument(text='hello world'), return_type=MyDocument ): print(doc.text)
🐞 Bug 修复
-
修复动态批处理的超时问题 (#6071)
无论是在执行器端点使用
@dynamic_batching
装饰器还是在部署中设置uses_dynamic_batching
参数,使用动态批处理时,由于计时器可能要等到前一个批处理完成后才触发,因此超时未被正确考虑。目前该问题已被修复。
-
修复不同容器化执行器中相同文档类型的问题 (#6062)
当 Flow 内的多个 Executor 具有相同的模型名称,并且它们用作 docker 容器或设置环境变量
JINA_MP_START_METHOD=spawn
时,通过 HTTP 提供服务时会出现问题。通过确保每个模型名称只提供一个模型实例,该问题已被修复。 -
修复拓扑结构模式验证 (#6057)
以前,如果 Flow 中不同 Executor 的端点模型模式不完全匹配,Flow 将无法启动。即使差异很小,小到只是不同的默认值,也会引发此错误。
我们通过放宽模型模式检查,仅验证属性类型是否匹配来修复此错误。
-
修复共识模块内存泄漏 (#6054)
在共识的 Golang 模块中,一些分配的字符串没有被正确释放。我们已经修复了这个问题。
-
在 Flow 网关中进行文档转换 (#6032)
这个错误与 #6027 有关。现在,我们在 GatewayStreamer 中使用
return_type
参数,以确保在网关级别接收的文档被转换为正确的模式。这可以防止先前发生的验证和序列化错误。 -
移除 Sandbox (#6047)
由于 Sandbox 已被弃用,因此不再支持在 Jina Cloud 沙盒中部署 Executor。
-
在注释中跳过文档属性,但不在字段中跳过 (#6035)
当在 Flow 中部署 BaseDoc 模型的 Executor 且该模型带有
ClassVar
值属性,由于 Gateway 无法正确创建模式,服务将无法初始化。我们通过在动态创建这些 Pydantic 模型时保护对__fields__
的访问来修复了这个问题。 -
使 Gateway 负载均衡流式结果 (#6024)
当在 Deployment 中使用
include_gateway=True
时,Executors 中的 endpoints 可以部署在 Gateway 后面。在这种情况下,Gateway 充当负载均衡器。在之前的版本中,当使用 HTTP 协议时,Gateway 会等待直到来自 Executor 的响应的所有分块都已经完成流式传输。
只有当接收到所有分块后,它才会将它们发送回客户端。这种处理方式导致了延迟,并抑制了流式 endpoints 的期望行为(即显示具有打字机效果的 LLM 流式传输的标记)。
这个版本修复了以上问题。Gateway接收到Executor响应的分块将立即进行流式传输。
无论您在Deployment中将
include_gateway
设置为True还是False,流式endpoints应该具有相同的期望行为。 -
修复Executors和Flows中深度嵌套模式的支持 (#6021)
当将深度嵌套模式(具有2级或更多级嵌套的DocArray模式)指定为Executor端点的输入或输出,并在Flow中部署了Executor时,Gateway将无法获取有关endpoints及其输入/输出模式的信息。
from typing import List, Optional from docarray import BaseDoc, DocList from jina import Executor, Flow, requests class Nested2Doc(BaseDoc): value: str class Nested1Doc(BaseDoc): nested: Nested2Doc class RootDoc(BaseDoc): nested: Optional[Nested1Doc] class NestedSchemaExecutor(Executor): @requests(on='/endpoint') async def endpoint(self, docs: DocList[RootDoc], **kwargs) -> DocList[RootDoc]: rets = DocList[RootDoc]() rets.append( RootDoc( text='hello world', nested=Nested1Doc(nested=Nested2Doc(value='test')) ) ) return rets flow = Flow().add(uses=NestedSchemaExecutor) with flow: res = flow.post( on='/endpoint', inputs=RootDoc(text='hello'), return_type=DocList[RootDoc] )
2023-08-07 02:49:32,529 topology_graph.py[608] WARNING Getting endpoints failed: 'definitions'. Waiting for another trial
这是由于内部效用程序函数无法将深度嵌套的 JSON 模式转换为 DocArray 模型。
新版本为嵌套模式生成模型时传播全局模式定义,修复了上述问题。
-
修复从所有endpoints、输入和输出缓存模型 (#6005)
修复了在Flow中使用Executor时的问题,该问题主要体现在相同的文档类型在不同endpoints中用作输入和输出。
-
将 127.0.0.1 用作本地 ctrl 地址 (#6004)
在本地工作时,编排层将使用 127.0.0.1 来发送对Executors和Gateways的健康检查。在以前的版本中,我们使用 0.0.0.0 作为默认主机,这在某些配置中会出现问题。
-
忽略来自 Google 的警告 (#5968)
使与 pkg_resources 弃用的 API 相关的警告减少出现。