evofabric.app.sop2workflow

evofabric.app.sop2workflow#

evofabric.app.sop2workflow.extract_text_between(text: str, start: str, end: str) → str | None[source]#

Extract the substring between the specified start and end markers from the given string.

Parameters:

  • text (str) – Original text.

  • start (str) – Start marker string

  • end (str) – End marker string.

Returns:

If the start and end markers are successfully found, return the substring between them; otherwise return None.

Return type:

Optional[str]

evofabric.app.sop2workflow.generate_condition_router_function_call(source: str, possible_targets: list, fallback_target: str = 'end', exit_function_name: str = None)[source]#

Generate a conditional routing function that can be injected into the system prompt of a decision node, used to determine the next node to execute based on the model’s output.

Parameters:

  • source (str) – The name of the current decision node.

  • possible_targets (list[str]) – List of target node names the current node may jump to.

  • fallback_target (str, optional) – The default node name to which the model redirects when it returns an unknown or empty selection. Default: end

  • exit_function_name (str | None, optional) – Optional parameter. If provided, the model can immediately jump to the end node (emergency exit) by calling the tool function with that name.

Returns:

A routing function that takes the state object state as input and returns the target node name.

Return type:

callable

evofabric.app.sop2workflow.user_feedback_router(state)[source]#

User feedback routing function. Find the last assistant message and redirect the flow back to the node corresponding to that message.

Parameters:

state (object) – Current state object, must include the messages attribute.

Returns:

Target node name, if not found, return end.

Return type:

str

class evofabric.app.sop2workflow.GraphDespNode(BaseModel)[source]#

Node description information, inherited from BaseModel.

Parameters:

  • name (str) – node name.

  • tools (List[str]) – List of tool names used by the node.

  • memories (List[str]) – List of memory names used by this node.

  • instruction (str) – The command content of this node.

  • sop (Optional[str], optional) – The Standard Operating Procedure (SOP) fragment used when building this node.

class evofabric.app.sop2workflow.GraphDespEdge(BaseModel)[source]#

Description information of edges in the diagram, inherits from BaseModel

Parameters:

  • source (str) – Name of the starting node of the edge.

  • possible_targets (List[str]) – The list of target node names that the edge may point to.

  • type (Literal["condition"], optional) – Edge type, default: condition.

class evofabric.app.sop2workflow.GraphDescription(BaseModel)[source]#

The description information of the entire graph structure, inherits from BaseModel.

Parameters:

  • nodes (List[GraphDespNode]) – List of all nodes in the figure.

  • edges (List[GraphDespEdge]) – List of all edges in the figure.

  • entry_point – Graph’s entry node name

  • global_instruction (str) – Global command shared by all nodes.

class evofabric.app.sop2workflow.WorkflowGeneratorBase(BaseComponent)[source]#

The base class for the workflow generator, inheriting from BaseComponent.

Parameters:

sop (str) – Standard Operating Procedure (SOP) for generating workflows.

generate(self) → GraphEngine[source]#

Use Standard Operating Procedure (SOP) to generate a runnable graph engine.

Returns:

Generated graph engine instance.

Return type:

GraphEngine

load_yaml(file_path) → Any[source]#

Static method. Loads the content of the YAML file at the specified path.

Parameters:

file_path (str) – YAML file path.

Returns:

Loaded YAML data, if the file does not exist, return None.

Return type:

Any

dump_yaml(data, file_path) → None[source]#

Static method. Write data to a YAML file at the specified path.

Parameters:

  • data (Any) – Data to be written

  • file_path (str) – YAML file path.

Returns:

No return value.

Return type:

None

class evofabric.app.sop2workflow.SopBreakdownNodeDesp(BaseModel)[source]#

Describe a decomposed SOP workflow node.

Parameters:

  • name (str) – node name.

  • type (Literal["sop", "connect"]) – Node type can only be sop or connect. Nodes of the sop type strictly enforce SOP fragments; connect type nodes are routing nodes used to connect various nodes.

  • duty (str) – This Node’s Responsibility Description

  • instruction (str) – This node’s execution instruction.

  • next_node_routing_rule (Dict[str, str]) – The routing rules of this node, where the key is the target node name and the value is the trigger condition.

to_full_instruction(self, global_instruction) → str#

Based on global instructions and the node’s own information, generate complete node execution instructions.

Parameters:

global_instruction (str) – Global execution strategy or explanation.

Returns:

Complete node execution instruction text.

Return type:

str

class evofabric.app.sop2workflow.SopBreakdownGraphDesp(BaseModel)[source]#

Describe a decomposed SOP workflow diagram structure.

Parameters:

  • nodes (List[SopBreakdownNodeDesp]) – Node list.

  • global_instruction (str) – Global command shared by all nodes.

  • entry_point (str) – Name of the workflow entry node.

class evofabric.app.sop2workflow.WorkflowGenerator(WorkflowGeneratorBase, BaseComponent)[source]#

A component that automatically generates workflow graph (Graph) based on SOP (Standard Operating Procedure), inheriting from WorkflowGeneratorBase and BaseComponent.

Parameters:

  • graph_generation_client (ChatClientBase) – Large model client for generating graph structures.

  • graph_node_complete_client (ChatClientBase) – Large model client for improving node information.

  • graph_run_client (ChatClientBase) – Client for running large models that generate images.

  • retry (int) – The number of retries when large model response parsing fails, default: 5.

  • output_dir (Optional[str]) – Directory for caching generated graph description files. If set to None, the files will be regenerated each time; if a directory path is provided, it will attempt to load existing files to skip the generation step.

  • tools (List[ToolManagerBase]) – List of tool managers available to nodes in the workflow.

  • memories (Dict[str, Tuple[str, MemBase]]) – The dictionary of memory modules accessible by nodes in a workflow, where keys are custom names and values are (description, instance) tuples.

  • state_schema (Optional[List[Tuple[str, Any, str]]]) – A list of field definitions to be added to the state in addition to the default messages field. Format: [(field name, field type, field description), ...].

  • addition_global_instruction (str) – Global instruction snippet added to each node’s system prompt.

  • user_node (Optional[AsyncNode]) – The reserved node used when a node needs to interact with the user.

  • fallback_node (Optional[str]) – The default jump node when the node jump target is invalid or missing.

  • auto_self_loop (bool) – Whether to allow nodes to jump to themselves by default.

  • sop_disassembly_prompt (str) – Prompt template for decomposing complete SOP into workflow nodes

  • node_completion_prompt (str) – Prompt template for refining each node’s information (e.g., tools, memory, instructions, etc.)

  • tool_list_mode (Literal["all", "select"]) – How the control node obtains the tool list, optional all (all) or select (select by LLM).

  • memory_list_mode (Literal["all", "select"]) – How the control node retrieves the memory module list, optional all (all) or select (select by LLM).

  • skeleton_file_name (str) – The filename for saving the graph description, default: _skeleton.yaml.

  • reserved_nodes (List[str]) – List of reserved node names prohibited from being generated by LLM, default is ["start", "end", "user"].

  • exit_function_name (Optional[str]) – The function name that immediately jumps to the end node when the tool is called.

  • build_kwargs (Dict[str, Any]) – Additional parameters passed to evofabric.core.graph.GraphBuilder.build()

generate(self) → GraphEngine[source]#

Asynchronously generate a complete workflow diagram engine object.

Returns:

Runnable GraphEngine instance.

Return type:

GraphEngine

Contents