promptflow.batch 模块#

class promptflow.batch.BatchEngine(flow_file: Union[Path, Callable], working_dir: Optional[Path] = None, *, connections: Optional[dict] = None, storage: Optional[AbstractRunStorage] = None, batch_timeout_sec: Optional[int] = None, line_timeout_sec: Optional[int] = None, worker_count: Optional[int] = None, init_kwargs: Optional[Dict[str, Any]] = None, **kwargs)#

基类：object

此类别用于以批处理模式执行流

cancel()#

取消批处理运行

classmethod register_executor(language: str, executor_proxy_cls: Type[AbstractExecutorProxy])#

为特定编程语言注册执行器代理类。

此函数用于保持与旧版 promptflow-runtime 的兼容性；它会将注册重定向到 ExecutorProxyFactory。

run(input_dirs: Dict[str, str], inputs_mapping: Dict[str, str], output_dir: Path, run_id: Optional[str] = None, max_lines_count: Optional[int] = None, raise_on_line_failure: Optional[bool] = False, resume_from_run_storage: Optional[AbstractBatchRunStorage] = None, resume_from_run_output_dir: Optional[Path] = None, executor_proxy: Optional[AbstractExecutorProxy] = None) → BatchResult#

以批处理模式运行流

参数:

  • input_dirs (Dict[str, str]) – 输入文件的目录路径

  • inputs_mapping (Dict[str, str]) – 输入名称与其对应值的映射。

  • output_dir (输出文件的目录路径) – 输出目录

  • run_id (Optional[str]) – 此运行的运行 ID

  • max_lines_count (Optional[int]) – 输入的最大计数。如果为 None，则将使用所有输入。

  • raise_on_line_failure (Optional[bool]) – 当行失败时是否引发异常。

  • resume_from_run_storage (Optional[AbstractRunStorage]) – 从原始运行加载流运行和节点运行的运行存储。恢复行为是重用原始运行中成功的行结果并运行/重新运行剩余/失败的行。

  • resume_from_run_output_dir (Optional[Path]) – 原始运行的输出目录。

返回:

此批处理运行的结果

返回类型:

BatchResult

class promptflow.batch.BatchResult(status: Status, total_lines: int, completed_lines: int, failed_lines: int, node_status: Mapping[str, int], start_time: datetime, end_time: datetime, metrics: Mapping[str, str], system_metrics: SystemMetrics, error_summary: ErrorSummary)#

基类：object

批处理运行的结果。

completed_lines: int#
classmethod create(start_time: datetime, end_time: datetime, line_results: List[LineResult], aggr_result: AggregationResult, status: Status = Status.Completed, exception: Optional[Exception] = None) → BatchResult#
end_time: datetime#
error_summary: ErrorSummary#
failed_lines: int#
metrics: Mapping[str, str]#
node_status: Mapping[str, int]#
start_time: datetime#
status: Status#
system_metrics: SystemMetrics#
total_lines: int#