agentscope.pipelines package
Submodules
Module contents
Import all pipeline related modules in the package.
- class agentscope.pipelines.PipelineBase[source]
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 agentscope.pipelines.SequentialPipeline(operators: Sequence[Operator])[source]
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 agentscope.pipelines.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>)[source]
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)
- __init__(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>) None [source]
Initialize an IfElsePipeline.
- Parameters:
condition_func (Callable[[dict], bool]) – A function that determines whether to execute if_body_operators or else_body_operators based on the input x.
if_body_operators (Operators) – Operators executed when condition_func returns True.
else_body_operators (Operators) – Operators executed when condition_func returns False, does nothing and just return the input by default.
- class agentscope.pipelines.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>)[source]
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)
- __init__(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>) None [source]
Initialize a SwitchPipeline.
- Parameters:
condition_func (Callable[[dict], Any]) – A function that determines which case_operator to execute based on the input x.
case_operators (dict[Any, Operators]) – 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.
- class agentscope.pipelines.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>>)[source]
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
- __init__(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>>)[source]
Initialize a ForLoopPipeline.
- Parameters:
loop_body_operators (Operators) – Operators executed as the body of the loop.
max_loop (int) – Maximum number of loop executions.
(`Callable[[dict] (break_func) –
bool]`
_ (defaults to `lambda) –
False`) – A function used to determine whether to break out of the loop based on the output of the loop_body_operators.
- class agentscope.pipelines.WhileLoopPipeline(loop_body_operators: ~agentscope.agents.operator.Operator | ~typing.Sequence[~agentscope.agents.operator.Operator], condition_func: ~typing.Callable[[int, dict], bool] = <function WhileLoopPipeline.<lambda>>)[source]
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
- __init__(loop_body_operators: ~agentscope.agents.operator.Operator | ~typing.Sequence[~agentscope.agents.operator.Operator], condition_func: ~typing.Callable[[int, dict], bool] = <function WhileLoopPipeline.<lambda>>)[source]
Initialize a WhileLoopPipeline.
- Parameters:
loop_body_operators (Operators) – Operators executed as the body of the loop.
(`Callable[[int (condition_func) –
dict]
bool]`
to (defaults)
_ (`lambda) –
False`): A function that determines whether to continue executing the loop body based on the current loop number and output of the loop_body_operator
__ – False`): A function that determines whether to continue executing the loop body based on the current loop number and output of the loop_body_operator
- agentscope.pipelines.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.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.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.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.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