evofabric.core.graph

Graph Builder

class evofabric.core.graph.GraphBuilder(BaseComponent)

State-driven graph builder for orchestrating the execution logic of nodes and edges, inheriting from BaseComponent.

Parameters:

state_schema (type[StateSchema]) – The state pattern used in this figure can be a subclass of dict or pydantic.BaseModel.

add_node(self, name: str, node: Callable | NodeBase, action_mode: NodeActionMode | str = 'any', multi_input_merge_strategy: dict[str, Callable[[List[State]], State]] = None)

Add a node to the graph.

Parameters:

• name (str) – Node names must be unique within the graph. Do not use start and end, as these are reserved special nodes by the graph engine.

• node (Union[Callable, NodeBase]) – Node execution body, can be a regular Python function or an instance of a subclass of NodeBase.

• action_mode (Union[NodeActionMode, str]) – Node Trigger Strategy. The default is any. "all" means triggering only after all predecessor nodes have completed; "any" means triggering as soon as any predecessor node completes.

• multi_input_merge_strategy (dict[str, Callable[[List[State]], State]]) – Specify the multi-predecessor state merging strategy by channel name, where the key is the channel name and the value is a callable object of type Callable[[List[State] -> State]].

add_edge(self, source: str, target: str, group: str = 'all', state_filter: Callable[[State], State] | None = None)

Establish an ordinary edge between two nodes.

Parameters:

• source (str) – Source node name.

• target (str) – Target node name.

• group (str) – Edge’s group name, used for batch control or visualization grouping.

• state_filter (Optional[Callable[[State], State]]) – Optional state filtering function that intercepts and rewrites the states passing through this edge.

add_condition_edge(self, source: str, router: Callable, possible_targets: List[str] | Set[str], group: str = 'all')

Add a conditional routing edge to the specified source node.

router supports parsing four types of return results:

• Single str: next hop target node name;

• List[str]: multiple next-hop target node names;

• Tuple[str, Callable]: target node name + edge-specific state filtering function;

• List[Tuple[str, Callable]]: Multiple groups (target node name, state filter function).

Parameters:

• source (str) – Source node name.

• router (Callable) – Routing function that determines the next hop based on runtime state, with input being the full State output by the source node. .. note:: Note: Each node’s output is incremental state information. However, the input to the router function is the complete state formed by merging the node’s incremental output with the input state.

• possible_targets (Union[List[str], Set[str]]) – All target node names that may be routed to, used for verification and optimization.

• group (str) – Edge’s group name.

set_entry_point(self, entry_name: str)

Set the graph’s unique entry node.

Parameters:

entry_name (str) – Ingress node name, must have been added via add_node.

build(self, auto_conn_end: bool = True, max_turn: int = None, timeout: int = None, graph_mode: GraphMode = GraphMode.RUN, db_file_path: str = './.storage.db', db_name: str = 'evofabric')

Construct an executable graph instance based on the added nodes and edges.

Parameters:

• auto_conn_end (bool) – If set to True, automatically connect all nodes without successors to the built-in END node.

• max_turn (int) – Maximum number of graph runs, prevent infinite loops.

• timeout (int) – Graph run total timeout time (seconds).

• graph_mode (GraphMode | str) – Graph run mode, default run means execution mode, debug means debug mode.

• db_file_path (str) – Persistent database file path, used for storing graph state snapshots in debug mode.

• db_name (str) – Database name.

Returns:

An executable GraphEngine or GraphEngineDebugger graph engine instance

dumps(self, graph_name: str = 'graph', version: str = '1.0') → dict

Serialize the nodes and edges in the current graph builder into a dictionary configuration.

Parameters:

• graph_name (str) – Graph name, default is “graph”.

• version (str) – Configuration version number, default is “1.0”.

Returns:

A dictionary containing the graph name, version, state mode, nodes, edges, and whether the entry point is set.

Return type:

dict

dump(self, save_path: str, graph_name: str = 'graph', version: str = '1.0')

Serialize and save the content of the current graph builder to the specified file.

Parameters:

• save_path (str) – Save path.

• graph_name (str) – Graph name, default is “graph”.

• version (str) – Configuration version number, default is “1.0”.

load(cls, file_path: str) → 'GraphBuilder'

Load the pre-built graph builder from the file.

Parameters:

file_path (str) – File path.

Returns:

The GraphBuilder instance after loading.

Return type:

GraphBuilder

loads(cls, data: dict) → 'GraphBuilder'

Load the pre-built graph builder from the configuration dictionary.

Parameters:

data (dict) – Dictionary containing graph configuration.

Returns:

The GraphBuilder instance after loading.

Return type:

GraphBuilder

Graph Engine

class evofabric.core.graph.GraphEngine(BaseComponent)

Graph Execution Engine, responsible for parsing and driving the execution of the entire graph (nodes + edges), supporting state checkpoints, maximum rounds, and timeout control.

Parameters:

• nodes (Dict[str, GraphNodeSpec]) – The specification mapping of all nodes in the diagram, where the keys are node names and the values are GraphNodeSpec objects.

• edges (Dict[str, List[EdgeSpecBase]]) – Mapping of all edge specifications in the graph, where keys are source node names and values are lists of EdgeSpecBase for edges originating from those nodes.

• state_schema (Optional[SkipValidation[type[StateSchema]]]) – State pattern type, used for runtime state validation; when set to None, no validation is performed.

• max_turn (Optional[int]) – Maximum allowed number of node calls; exceeding this will forcibly terminate the running graph, but outputs that have reached the END node are still retained; a value of None indicates no limit.

• timeout (Optional[int]) – Timeout duration for a single node execution (seconds); setting to None means no limit.

async run(self, inputs: Dict)

Asynchronously start graph execution.

Parameters:

inputs (Dict) – Initial input data will be injected into the state of the start node, with keys and values strictly corresponding to StateSchema.

Returns:

Final state after the run ends.

Return type:

Dict

draw_graph(self, save_path: str = None, auto_open: bool = True)

Generate a graph visualization HTML file.

Parameters:

• save_path (str) – File save path; when it is None, do not persist to disk.

• auto_open (bool) – Open in browser automatically after generation.

class evofabric.core.graph.RunTimeTask(BaseModel)

Used to describe node tasks that need to be scheduled during graph execution and their related states and filtering conditions.

Parameters:

• node_name (str) – Target node name.

• state_ckpt (Union[StateCkpt, List[StateCkpt]]) – The StateCkpt corresponding to the current task; supports a single StateCkpt or a list.

• edge_group (str) – The group name of the edge to which the current task belongs, used to distinguish different edge sets in parallel or conditional branches.

• predecessor (Optional[Union[str, List[str]]]) – Predecessor node name; supports a single node name or a list of node names, with None indicating no predecessor node.

• state_filter (Optional[Callable]) – The state filter function applied to edges must have a signature of Callable[[Any], bool]; if None, no additional filtering is performed.

• trace_route (List[str]) – Executed node trajectory, recording the node names in order along the path; default empty list.

Graph Engine Debugger

class evofabric.core.graph.GraphEngineDebugger(GraphEngine)

Debuggable version of GraphEngine, supporting breakpoints, step execution, state restoration, and node output modification.

Parameters:

• db_file_path (str) – Database file path for state storage. Default value ".state_storage.db"

• db_name (str) – Database name. Default value "graph state"

model_post_init(context: Any) → None

Initialize the internal structure after model creation.

reset() → None

Reset all internal states, including tracking tree and breakpoints.

set_breakpoint(node_name_bp: str | None = None, condition_bp: Any = None, condition: Any = None) → None

Set a breakpoint at the specified node.

Parameters:

• node_name_bp (Optional[str]) – The node name to set a breakpoint on.

• condition_bp (Any) – Breakpoint Condition (Temporarily Not Supported)

• condition (Any) – Breakpoint Condition (Temporarily Not Supported)

Raises:

RuntimeError – Raises an exception when node_name_bp is None.

clear_breakpoint(node_name_bp: str | None = None, condition_bp: Any = None, condition: Any = None) → None

Clear the breakpoint at the specified node.

Parameters:

• node_name_bp (Optional[str]) – Node name to clear the breakpoint.

• condition_bp (Any) – Breakpoint Condition (Temporarily Not Supported)

• condition (Any) – Breakpoint Condition (Temporarily Not Supported)

Raises:

RuntimeError – Raises an exception when node_name_bp is None.

clear_all_breakpoint() → None

Clear all breakpoints.

async resume(running_queue: List[RunTimeTask] | None = None) → Awaitable[Any]

Continue from the current breakpoint or the specified RuntimeTask queue.

Parameters:

running_queue (Optional[List[RunTimeTask]]) – Pending task queue; if None, restore from non-checkpoint leaf nodes.

Returns:

Single-step execution result.

Return type:

Any

Raises:

RuntimeError – Thrown when the graph engine has already been started.

async step_over(node_uuid: str | None = None) → Awaitable[Any]

Skip the current breakpoint or specified node.

Parameters:

node_uuid (Optional[str]) – UUID of the node to skip; if None, skip all current nodes.

Returns:

Single-step execution result.

Return type:

Any

restore_step(node_uuid: str | None = None) → None

Undo the last operation.

Parameters:

node_uuid (Optional[str]) – The UUID of the node to be restored; if None, restore the previous step.

async run_one_step(running_queue: List[RunTimeTask]) → Awaitable[Tuple[Any, List[RunTimeTask]]]

Execute a step in the diagram.

Parameters:

running_queue (List[RunTimeTask]) – Pending Task Queue.

Returns:

Output results and the next batch of candidate task nodes.

Return type:

Tuple[Any, List[RunTimeTask]]

async debug(inputs: Dict) → Awaitable[Any]

Start the debugging session and inject the initial input.

Parameters:

inputs (Dict) – Initial state dictionary.

Returns:

Final result after debugging

Return type:

Any

change_output(from_node_uuid: str, to_node_uuid: str, change_key: str, change_value: Any) → None

Modify the output value between nodes.

Parameters:

• from_node_uuid (str) – Source Node UUID.

• to_node_uuid (str) – Target Node UUID.

• change_key (str) – Key to be modified

• change_value (Any) – New key-value.

Raises:

RuntimeError – Throw when the target node is not the current leaf node.

Example

from typing import Annotated, List, TypedDict
from pydantic import BaseModel
from evofabric.core.graph import GraphBuilder
from evofabric.core.typing import *
from evofabric.logger import get_logger

class State(TypedDict):
    messages: Annotated[List, "append_messages"]
    node_output: Annotated[str, "overwrite"]

def node_a(state):
    return {"messages": [AssistantMessage(content="node a output")], 'node_output': 'a'}

def node_b(state):
    self.assertIsInstance(state, dict)
    self.assertIsInstance(state['messages'][0], StateMessage)
    return {"messages": [AssistantMessage(content="node b output")], 'node_output': 'b'}

def node_c(state):
    self.assertIsInstance(state, dict)
    self.assertIsInstance(state['messages'][0], StateMessage)
    return {"messages": [AssistantMessage(content="node c output")], 'node_output': 'c'}

def node_d(state):
    self.assertIsInstance(state, dict)
    self.assertIsInstance(state['messages'][0], StateMessage)
    return {"messages": [AssistantMessage(content="node d output")], 'node_output': 'd'}

graph_builder = GraphBuilder(state_schema=State)
graph_builder.add_node("a", node_a)
graph_builder.add_node("b", node_b)
graph_builder.add_node("c", node_c)
graph_builder.add_node("d", node_d)
graph_builder.add_edge("a", "b")
graph_builder.add_edge("b", "c")
graph_builder.add_edge("c", "d")
graph_builder.set_entry_point("a")
graph = graph_builder.build(graph_mode=GraphMode.DEBUG, db_file_path="D:/Download/.db_storage.db")

# debug mode example
graph.set_breakpoint(node_name_bp="b")
await graph.debug({"messages": [UserMessage(content='hello')]})
await graph.step_over(node_uuid=graph._trace_tree.get_leaf_node_uuid_by_node_name(node_name="b"))
graph.restore_step()
graph.clear_breakpoint(node_name_bp="b")
result = await graph.resume()

Node

class evofabric.core.graph.NodeBase(BaseComponent)

Abstract base class for all nodes, inheriting from BaseComponent. Subclasses must implement the __call__ protocol to define the specific behavior of the node when called.

class evofabric.core.graph.SyncNode(NodeBase)

Synchronous execution node, inheriting from NodeBase. When called, it receives the current state synchronously and returns the state increment.

__call__(self, state: State) → StateDelta

Synchronous execution of node logic.

Parameters:

state (State) – Current workflow status.

Returns:

The amount of change relative to the current state.

Return type:

StateDelta

class evofabric.core.graph.AsyncNode(NodeBase)

Asynchronous Execution Node, inheriting from NodeBase. When called, it receives the current state asynchronously and returns a state increment.

async __call__(self, state: State) → StateDelta

Asynchronous execution of node logic.

Parameters:

state (State) – Current workflow status.

Returns:

The amount of change relative to the current state.

Return type:

StateDelta

class evofabric.core.graph.SyncStreamNode(NodeBase)

Nodes supporting synchronous streaming output inherit from NodeBase. During the call, streaming data blocks can be sent in real time via stream_writer.

__call__(self, state: State, stream_writer: StreamWriter) → StateDelta

Synchronous streaming execution node logic.

Parameters:

• state (State) – Current workflow status.

• stream_writer (StreamWriter) – Writer for real-time output of streaming data.

Returns:

The amount of change relative to the current state.

Return type:

StateDelta

class evofabric.core.graph.AsyncStreamNode(NodeBase)

Nodes supporting asynchronous streaming output inherit from NodeBase. During the call, streaming data blocks can be sent in real-time via stream_writer.

async __call__(self, state: State, stream_writer: StreamWriter) → StateDelta

Asynchronous streaming execution node logic.

Parameters:

• state (State) – Current workflow status.

• stream_writer (StreamWriter) – Writer for real-time output of streaming data.

Returns:

The amount of change relative to the current state.

Return type:

StateDelta

class evofabric.core.graph.GraphNodeSpec(BaseComponent)

This node is used in the graph engine, first injecting graph runtime information into the context, then determining the invocation method based on different node types and distributing different input parameters. Inherited from BaseComponent.

Parameters:

• node (NodeBase) – Node instance must inherit from NodeBase.

• node_name (str) – The human-readable name of the node in the graph.

• action_mode (NodeActionMode) – Node execution mode, default NodeActionMode.ALL.

• stream_writer (Optional[StreamWriter]) – Optional streaming write pointer, used to push the node’s intermediate results in real-time.

• multi_input_merge_strategy (Optional[Dict[str, Callable[[List[State]], State]]]) – When a node has multiple inputs, specify a custom merge function by key Callable[[List[State]], State]; if not provided, use the default strategy.

• node_id (str) – Node global unique identifier, default auto-generated UUID.

is_active(self) → bool

@property Type. Whether the current node is active.

Returns:

Returns True if activated, otherwise False.

Return type:

bool

async __call__(self, state: State, **kwargs) → StateDelta

Inject node execution information via evofabric.core.graph.stream_writer_env() and execute node logic under asynchronous lock protection.

Parameters:

state (State) – Input status object.

Returns:

The state increment generated after node execution.

Return type:

StateDelta

evofabric.core.graph.callable_to_node(callable_obj: Callable[..., Any]) → NodeBase

Automatically convert any callable object (function or class instance that implements the __call__ method) to the corresponding node type instance.

The function will automatically determine its type based on the defining characteristics of the target object:

• If it is an asynchronous function and has the stream_writer parameter → convert to asynchronous streaming node.

• If it is an asynchronous function and does not have the stream_writer parameter → convert to asynchronous ordinary node.

• If it is a synchronous function and has the stream_writer parameter → convert to a synchronous streaming node.

• If it is a synchronous function and does not have the stream_writer parameter → convert to synchronous ordinary node.

Parameters:

callable_obj (Callable[..., Any]) – Any callable object, which can be a function, a method, or a class instance that implements __call__.

Returns:

The corresponding node type instance, whose type is a subclass of NodeBase.

Return type:

NodeBase

Edge

class evofabric.core.graph.EdgeSpecBase(ABC, BaseComponent)

Abstract base class for edge specifications, defining the basic properties and interfaces of edges in a graph. Inherits from ABC and BaseComponent.

Parameters:

• source (str) – Source node name.

• group (str) – The group the edge belongs to, default is DEFAULT_EDGE_GROUP.

• edge_type (Literal['base']) – Edge type identifier, default is "base".

get_targets(self, state: State) → List[Tuple[str, StateFilterLike | None]]

Abstract method. Returns the target node for the next step and its corresponding state filter based on the current state.

Parameters:

state (State) – Current graph status.

Returns:

List of target nodes and state filters.

Return type:

List[Tuple[str, Optional[StateFilterLike]]]

get_possible_targets(self) → List[str]

Abstract method. Returns all target nodes that this edge may reach in static analysis.

Returns:

All possible target node names.

Return type:

List[str]

class evofabric.core.graph.EdgeSpec(EdgeSpecBase)

Normal edge type, connects only one target node. Inherits from EdgeSpecBase.

Parameters:

• source (str) – Source node name.

• group (str) – The group the edge belongs to, default is DEFAULT_EDGE_GROUP.

• edge_type (Literal['edge']) – Edge type identifier, fixed as "edge".

• target (str) – Target node name.

• state_filter (Optional[StateFilterLike]) – State filtering function, optional, used to control the trigger conditions of edges.

get_targets(self, state: State) → List[Tuple[str, StateFilterLike | None]]

Return the target node and its status filter, compatible with the unified interface.

Parameters:

state (State) – Current status.

Returns:

List of (target node, status filter).

Return type:

List[Tuple[str, Optional[StateFilterLike]]]

get_possible_targets(self) → List[str]

Return all possible target nodes for the current edge (a single node for ordinary edges).

Returns:

Target node name list.

Return type:

List[str]

class evofabric.core.graph.ConditionEdgeSpec(EdgeSpecBase)

Conditional edge type, supports dynamically determining the target node based on state. Inherits from EdgeSpecBase.

Parameters:

• source (str) – Source node name.

• group (str) – The group the edge belongs to, default is DEFAULT_EDGE_GROUP.

• edge_type (Literal['conditional']) – Edge type identifier, fixed as "conditional".

• router (Callable[[State], Union[str, List[str], Tuple[str, Callable], List[Tuple[str, Callable]]]]) – Routing function, accepts State as input, returns target node or node list, may include a state filtering function.

• possible_targets (List[str]) – List of all allowed target nodes.

get_targets(self, state: State) → List[Tuple[str, StateFilterLike | None]]

Call the router function to obtain the next target node based on the current state, and uniformly format it as [(node name, state filter function)].

Parameters:

state (State) – Current status.

Returns:

List of target nodes and state filtering functions.

Return type:

List[Tuple[str, Optional[StateFilterLike]]]

get_possible_targets(self) → List[str]

Return all possible target nodes of conditional edges under static analysis.

Returns:

All possible target node names.

Return type:

List[str]

evofabric.core.graph.cast_edge(v: dict) → EdgeSpecBase

Automatically determine the edge type based on the edge_type field in the input dictionary and deserialize it into the corresponding EdgeSpec or ConditionEdgeSpec instance.

Parameters:

v (dict) – Dictionary object containing edge definitions.

Returns:

Parsed edge object.

Return type:

EdgeSpecBase

State & Update

class evofabric.core.graph.StateUpdater

Class for registering and managing state update strategies.

Through this class, custom state merge functions can be registered as named strategies and retrieved by name when needed.

Usage Example:

@StateUpdater.register("overwrite")
def overwrite(old: Any, new: Any) -> Any:
    return new

strategy = StateUpdater.get("overwrite")
merged = strategy(old_state, new_state)

register(cls, name: str) → Callable[[Callable], Callable]

Class method for registering a new state update strategy.

Parameters:

name (str) – Strategy Name, after registration, the corresponding strategy function can be obtained through this name.

Returns:

A decorator function used to decorate the actual state update function.

Return type:

Callable[[Callable], Callable]

Raises:

KeyError – If the strategy name already exists, throw an exception.

get(cls, name: str) → Callable[[Any, Any], Any]

Class method, retrieves the registered state update strategy function by name.

Parameters:

name (str) – Strategy Name.

Returns:

The corresponding state update function accepts the old state and the new state as parameters and returns the merged state.

Return type:

Callable[[Any, Any], Any]

Raises:

KeyError – If the strategy name is not registered, throw an exception.

list_strategies(cls) → List[str]

Class method, returns a list of all registered policy names.

Returns:

List of all strategy names

Return type:

List[str]

registered(cls, name: str) → bool

Class method, determines whether a strategy with the specified name has been registered.

Parameters:

name (str) – Strategy Name.

Returns:

If the strategy is registered, return True; otherwise, return False.

Return type:

bool

class evofabric.core.graph.StateCkpt(BaseComponent)

Represents a state checkpoint inheriting from BaseComponent, used for managing states and their changes.

Parameters:

• delta (Optional[SkipValidation[StateDelta]]) – Change in state (increment), optional.

• parent (Optional[StateCkpt]) – Parent node, used for building a state chain, optional.

• state_schema (Optional[type[StateSchema]]) – Type definition of the state structure, optional.

• materialized_state_cache (Optional[SkipValidation[State]]) – The specific state of the cache, which defaults to None upon initialization and is not included in the initialization parameters.

materialize(self) → State | StateSchema

Recursively backtrack to parent nodes, merge state increments, obtain the complete State state.

Returns:

Concrete state object, type State or StateSchema.

Return type:

Union[State, StateSchema]

merge(cls, checkpoints: List['StateCkpt'], strategy: Callable[[List[State]], State] = None)

Merge multiple states into a new state.

Parameters:

• checkpoints (List[StateCkpt]) – List of states to be merged.

• strategy (Callable[[List[State]], State]) – Merge strategy function, receives multiple states and returns the merged state. If not provided, use the default logic declared in StateSchema.

Returns:

New State After Merging

Return type:

StateCkpt

filter(cls, checkpoint: 'StateCkpt', strategy: Callable[[State], State])

Apply the filtering strategy to the given state to generate a new state.

Parameters:

• checkpoint (StateCkpt) – Statuses that need to be filtered.

• strategy (Callable[[State], State]) – Filtering strategy function, receives a state and returns the processed state.

Returns:

New state checkpoint after applying the filtering strategy.

Return type:

StateCkpt

merge_state(state, delta, state_schema) → Dict | StateSchema

Static method used to merge state and increment into a new state.

Parameters:

• state (State) – Original state.

• delta (StateDelta) – State increment.

• state_schema (type[StateSchema]) – State structure definition, used for type conversion.

Returns:

The new state after merging, type is dictionary or StateSchema.

Return type:

Union[Dict, StateSchema]

evofabric.core.graph.generate_state_schema(variables: List[Tuple[str, Any, str]] | None = None)

Declare the variable names and types for the state information passed within the graph engine.

Precautions:

• Variable names must conform to Python’s variable naming conventions.

• The variable type must be one of the following: str, int, float, list, tuple, dict.

• There exists a constant variable named messages used to record agent context; please avoid defining other variables with the same name.

Example:

generate_state_schema([
    ("msg_id", str, "overwrite"),
    ("user_id", bool, "overwrite")
])

Parameters:

variables (Optional[List[Tuple[str, Any, str]]]) – A list of tuples, each containing variable name, variable type, and update strategy.

Returns:

The Pydantic model class dynamically generated based on the provided variables, represents the state structure.

Return type:

pydantic.BaseModel

Raises:

• ValueError – If the variable type is invalid.

• ValueError – If the variable name conflicts with a reserved field or is redefined.

• ValueError – If the update strategy is not registered in StateUpdater.

evofabric.core.graph._overwrite_state_update_strategy(old: Any = MISSING, new: Any = MISSING) → Any

The specific implementation of the update strategy overwrite.

The strategy function that overwrites the old value with the new value.

If the new value is MISSING, return the old value;

Parameters:

• old (Any) – Old state value, default is MISSING.

• new (Any) – New status value, default is MISSING.

Returns:

Updated status value.

Return type:

Any

evofabric.core.graph._append_messages(old: List[StateMessage] = MISSING, new: List[StateMessage] = MISSING) → List[StateMessage]

Specific implementation of the update strategy append_messages.

The strategy function that appends the new message list to the end of the old message list, automatically deduplicates.

If the old value or new value is MISSING, it is considered an empty list.

Parameters:

• old (List[StateMessage]) – Old Message List, default is MISSING.

• new (List[StateMessage]) – New message list, default is MISSING.

Returns:

Merged message list, deduplicated.

Return type:

List[StateMessage]

Streaming & Context

class evofabric.core.graph.StreamWriter(BaseComponent)

Stream Message Writer, used for sending message chunks during asynchronous streaming processing. This class inherits from BaseComponent.

put(payload: Any) → None

Place the streaming message content into the writer and trigger the registered message callback function.

Parameters:

payload (Any) – The streaming data content to be sent can be any type.

Returns:

No return value

Return type:

None

class evofabric.core.graph.StreamCtx(BaseModel)

Data model representing stream processing context information, inheriting from BaseModel.

Parameters:

• node_name (Optional[str]) – Current node name (optional).

• call_id (Optional[str]) – Current Call ID (Optional)

• tool_name (Optional[str]) – Current tool name (optional)

• tool_call_id (Optional[str]) – Current tool call ID (optional).

__bool__(self) → bool

Determine if the current context is empty. If all attributes are empty, return False, otherwise return True.

Returns:

Is there valid information in the context?

Return type:

bool

__repr__(self) → str

Returns the string representation of the current context.

Returns:

String describing current context

Return type:

str

evofabric.core.graph.set_streaming_handler(callback: Callable[[dict], None | Awaitable[None]]) → None

Register a callback function for processing streaming messages.

Usage Example:

def print_handler(payload):
    print(payload)

set_streaming_handler(print_handler)

Parameters:

callback (Callable[[dict], Union[None, Awaitable[None]]]) – Set the callback function for processing streaming messages, supporting synchronous and asynchronous types.

Returns:

No return value

Return type:

None

evofabric.core.graph.current_ctx() → StreamCtx

Obtain the stream processing context in the current thread/coroutine.

Returns:

Current Stream Processing Context Object

Return type:

StreamCtx

evofabric.core.graph.stream_writer_env(ctx_updates: StreamCtx)

Create a temporary stream processing context, and execute the code block within this context.

Usage Example:

def node(stream_writer):
    stream_writer.put("streaming msg 1")
    stream_writer.put("streaming msg 2")
    ...

with stream_writer_env(StreamCtx(call_id=str(uuid.uuid4()), node_name=self.node_name)):
    node(get_stream_writer())

Parameters:

ctx_updates (StreamCtx) – New field value for updating the current context.

Returns:

Context manager, used with the with statement.

Return type:

contextmanager

evofabric.core.graph.get_stream_writer()

Get a globally shared StreamWriter instance.

Returns:

Global StreamWriter instance

Return type:

StreamWriter