agentscope.pipelines package

Submodules

• agentscope.pipelines.functional module
  • forlooppipeline()
  • ifelsepipeline()
  • placeholder()
  • sequentialpipeline()
  • switchpipeline()
  • whilelooppipeline()
• agentscope.pipelines.pipeline module
  • ForLoopPipeline
  • IfElsePipeline
  • PipelineBase
  • SequentialPipeline
  • SwitchPipeline
  • WhileLoopPipeline

Module contents

Import all pipeline related modules in the package.

class ForLoopPipeline(loop_body_operators: ~agentscope.agents.operator.Operator | ~typing.Sequence[~agentscope.agents.operator.Operator], max_loop: int, break_func: ~typing.Callable[[dict], bool] = <function ForLoopPipeline.<lambda>>)

Bases: PipelineBase

A template pipeline for implementing control flow like for-loop

ForLoopPipeline(loop_body_operators, max_loop) represents the following workflow:

for i in range(max_loop):
    x = loop_body_operators(x)

ForLoopPipeline(loop_body_operators, max_loop, break_func) represents the following workflow:

for i in range(max_loop):
    x = loop_body_operators(x)
    if break_func(x):
        break

class IfElsePipeline(condition_func: ~typing.Callable[[dict], bool], 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>)

Bases: PipelineBase

A template pipeline for implementing control flow like if-else.

IfElsePipeline(condition_func, if_body_operators, else_body_operators) represents the following workflow:

if condition_func(x):
    if_body_operators(x)
else:
    else_body_operators(x)

class PipelineBase

Bases: Operator

Base interface of all pipelines.

The pipeline is a special kind of operator that includes multiple operators and the interaction logic among them.

class SequentialPipeline(operators: Sequence[Operator])

Bases: PipelineBase

A template pipeline for implementing sequential logic.

Sequential(operators) represents the following workflow:

x = operators[0](x)
x = operators[1](x)
...
x = operators[n](x)

class SwitchPipeline(condition_func: ~typing.Callable[[dict], ~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>)

Bases: PipelineBase

A template pipeline for implementing control flow like switch-case.

SwitchPipeline(condition_func, case_operators, default_operators) represents the following workflow:

switch condition_func(x):
    case k1: return case_operators[k1](x)
    case k2: return case_operators[k2](x)
    ...
    default: return default_operators(x)

class WhileLoopPipeline(loop_body_operators: ~agentscope.agents.operator.Operator | ~typing.Sequence[~agentscope.agents.operator.Operator], condition_func: ~typing.Callable[[int, dict], bool] = <function WhileLoopPipeline.<lambda>>)

Bases: PipelineBase

A template pipeline for implementing control flow like while-loop

WhileLoopPipeline(loop_body_operators, condition_operator, condition_func) represents the following workflow:

i = 0
while (condition_func(i, x))
    x = loop_body_operators(x)
    i += 1

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

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

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

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

sequentialpipeline(operators: Sequence[Operator], x: dict | None = None) → dict

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

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

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

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

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