agentscope.pipelines.functional module

Functional counterpart for Pipeline

agentscope.pipelines.functional.placeholder(x: dict | None = None) → dict[source]

A placeholder that do nothing.

Acts as a placeholder in branches that do not require any operations in flow control like if-else/switch

agentscope.pipelines.functional.sequentialpipeline(operators: Sequence[Operator], x: dict | None = None) → dict[source]

Functional version of SequentialPipeline.

Parameters:

  • operators (Sequence[Operator]) – Participating operators.

  • x (Optional[dict], defaults to None) – The input dictionary.

Returns:

the output dictionary.

Return type:

dict

agentscope.pipelines.functional.ifelsepipeline(condition_func: ~typing.Callable, if_body_operators: ~agentscope.agents.operator.Operator | ~typing.Sequence[~agentscope.agents.operator.Operator], else_body_operators: ~agentscope.agents.operator.Operator | ~typing.Sequence[~agentscope.agents.operator.Operator] = <function placeholder>, x: dict | None = None) → dict[source]

Functional version of IfElsePipeline.

Parameters:

  • condition_func (Callable) – A function that determines whether to exeucte if_body_operator or else_body_operator based on x.

  • if_body_operator (Operators) – Operators executed when condition_func returns True.

  • else_body_operator (Operators, defaults to placeholder) – Operators executed when condition_func returns False, does nothing and just return the input by default.

  • x (Optional[dict], defaults to None) – The input dictionary.

Returns:

the output dictionary.

Return type:

dict

agentscope.pipelines.functional.switchpipeline(condition_func: ~typing.Callable[[~typing.Any], ~typing.Any], case_operators: ~typing.Mapping[~typing.Any, ~agentscope.agents.operator.Operator | ~typing.Sequence[~agentscope.agents.operator.Operator]], default_operators: ~agentscope.agents.operator.Operator | ~typing.Sequence[~agentscope.agents.operator.Operator] = <function placeholder>, x: dict | None = None) → dict[source]

Functional version of SwitchPipeline.

Parameters:

  • condition_func (Callable[[Any], Any]) – A function that determines which case_operator to execute based on the input x.

  • case_operators (Mapping[Any, Operator]) – A dictionary containing multiple operators and their corresponding trigger conditions.

  • default_operators (Operators, defaults to placeholder) – Operators that are executed when the actual condition do not meet any of the case_operators, does nothing and just return the input by default.

  • x (Optional[dict], defaults to None) – The input dictionary.

Returns:

the output dictionary.

Return type:

dict

agentscope.pipelines.functional.forlooppipeline(loop_body_operators: ~agentscope.agents.operator.Operator | ~typing.Sequence[~agentscope.agents.operator.Operator], max_loop: int, break_func: ~typing.Callable[[dict], bool] = <function <lambda>>, x: dict | None = None) → dict[source]

Functional version of ForLoopPipeline.

Parameters:

  • loop_body_operators (Operators) – Operators executed as the body of the loop.

  • max_loop (int) – maximum number of loop executions.

  • break_func (Callable[[dict], bool]) – A function used to determine whether to break out of the loop based on the output of the loop_body_operator, defaults to lambda _: False

  • x (Optional[dict], defaults to None) – The input dictionary.

Returns:

The output dictionary.

Return type:

dict

agentscope.pipelines.functional.whilelooppipeline(loop_body_operators: ~agentscope.agents.operator.Operator | ~typing.Sequence[~agentscope.agents.operator.Operator], condition_func: ~typing.Callable[[int, ~typing.Any], bool] = <function <lambda>>, x: dict | None = None) → dict[source]

Functional version of WhileLoopPipeline.

Parameters:

  • loop_body_operators (Operators) – Operators executed as the body of the loop.

  • condition_func (Callable[[int, Any], bool], optional) – A function that determines whether to continue executing the loop body based on the current loop number and output of the loop_body_operator, defaults to lambda _,__: False

  • x (Optional[dict], defaults to None) – The input dictionary.

Returns:

the output dictionary.

Return type:

dict