agentscope.web.workstation.workflow_dag module

AgentScope workstation DAG running engine.

This module defines various workflow nodes that can be used to construct a computational DAG. Each node represents a step in the DAG and can perform certain actions when called.

agentscope.web.workstation.workflow_dag.remove_duplicates_from_end(lst: list) → list

remove duplicates element from end on a list

class agentscope.web.workstation.workflow_dag.ASDiGraph(*args, **kwargs)

基类：DiGraph

A class that represents a directed graph, extending the functionality of networkx’s DiGraph to suit specific workflow requirements in AgentScope.

This graph supports operations such as adding nodes with associated computations and executing these computations in a topological order.

nodes_not_in_graph

A set of nodes that are not included in

Type:

set

the computation graph.

__init__(*args, **kwargs)

Initialize the ASDiGraph instance.

run() → None

Execute the computations associated with each node in the graph.

The method initializes AgentScope, performs a topological sort of the nodes, and then runs each node’s computation sequentially using the outputs from its predecessors as inputs.

compile(compiled_filename: str = '', **kwargs) → str

Compile DAG to a runnable python code

add_as_node(node_id: str, node_info: dict, config: dict) → Any

Add a node to the graph based on provided node information and configuration.

参数:

• node_id (str) – The identifier for the node being added.

• node_info (dict) – A dictionary containing information about the node.

• config (dict) – Configuration information for the node dependencies.

返回:

The computation object associated with the added node.

exec_node(node_id: str, x_in: Any | None = None) → Any

Execute the computation associated with a given node in the graph.

参数:

• node_id (str) – The identifier of the node whose computation is to be executed.

• x_in – The input to the node’s computation. Defaults to None.

返回:

The output of the node’s computation.

agentscope.web.workstation.workflow_dag.sanitize_node_data(raw_info: dict) → dict

Clean and validate node data, evaluating callable expressions where necessary.

Processes the raw node information, removes empty arguments, and evaluates any callable expressions provided as string literals.

参数:

raw_info (dict) – The raw node information dictionary that may contain callable expressions as strings.

返回:

The sanitized node information with callable expressions

evaluated.

返回类型:

dict

agentscope.web.workstation.workflow_dag.build_dag(config: dict) → ASDiGraph

Construct a Directed Acyclic Graph (DAG) from the provided configuration.

Initializes the graph nodes based on the configuration, adds model nodes first, then non-model nodes, and finally adds edges between the nodes.

参数:

config (dict) – The configuration to build the graph from, containing node info such as name, type, arguments, and connections.

返回:

The constructed directed acyclic graph.

返回类型:

ASDiGraph

抛出:

ValueError – If the resulting graph is not acyclic.