Jina 近期更新

Jina 近期更新

🆕 新功能

• 优化动态批处理处理方式 (#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')
  1

• 参数可以描述为 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)
  1

• 使用 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()
  1

  在客户端中，可以使用新的 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)
  1

🐞 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]
      )
  1

  2023-08-07 02:49:32,529 topology_graph.py[608] WARNING Getting endpoints failed: 'definitions'. Waiting for another trial
  1

  这是由于内部效用程序函数无法将深度嵌套的 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 相关的警告减少出现。