plexus.scores.nodes.BaseNode module

class plexus.scores.nodes.BaseNode.BaseNode(**parameters)

Bases: ABC, LangChainUser

Abstract base class for nodes in a LangGraph workflow.

This class serves as a template for creating nodes that can be slotted into the main orchestration workflow of a LangGraphScore. Subclasses must implement the build_compiled_workflow method to define their specific behavior within the larger graph structure.

class GraphState(*, text: str, metadata: dict | None = None, results: dict | None = None, messages: ~typing.List[~typing.Dict[str, ~typing.Any]] | None = None, is_not_empty: bool | None = None, value: str | None = None, explanation: str | None = None, reasoning: str | None = None, chat_history: ~typing.List[~typing.Any] = <factory>, completion: str | None = None, classification: str | None = None, confidence: float | None = None, retry_count: int | None = 0, at_llm_breakpoint: bool | None = False, good_call: str | None = None, good_call_explanation: str | None = None, non_qualifying_reason: str | None = None, non_qualifying_explanation: str | None = None, **extra_data: ~typing.Any)

Bases: GraphState

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'allow', 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class Parameters(*, model_provider: Literal['ChatOpenAI', 'AzureChatOpenAI', 'BedrockChat', 'ChatVertexAI'] = 'AzureChatOpenAI', model_name: str | None = None, base_model_name: str | None = None, model_region: str | None = None, temperature: float | None = 0, top_p: float | None = 0.03, max_tokens: int | None = 500, input: dict | None = None, output: dict | None = None, system_message: str | None = None, user_message: str | None = None, example_refinement_message: str | None = None, single_line_messages: bool = False, name: str | None = None)

Bases: Parameters

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

example_refinement_message: str | None

input: dict | None

model_config: ClassVar[ConfigDict] = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None

output: dict | None

single_line_messages: bool

system_message: str | None

user_message: str | None

__init__(**parameters)

abstract add_core_nodes(workflow: StateGraph) → StateGraph

Build and return a core LangGraph workflow. The node name is available as self.node_name when needed.

build_compiled_workflow(graph_state_class: Type[GraphState]) → StateGraph

Build and return a compiled LangGraph workflow.

This method must be implemented by all subclasses to define the specific sub-graph that this node represents in the larger workflow.

Returns:

StateGraph: A compiled LangGraph workflow that can be used as a

sub-graph node in the main LangGraphScore workflow.

get_example_refinement_template()

Get the example refinement message for the node.

get_prompt_templates()

Get a list of prompt templates for the node, by looking for the “system_message” parameter for a system message, and the “user_message” parameter for a human message. If either is missing or empty, the method will skip that template.

log_state(state, input_state=None, output_state=None, node_suffix='')

Log the current state to the trace.node_results list in the state’s metadata.

Parameters

stateGraphState

The current state object

input_statedict, optional

The input state to log

output_statedict, optional

The output state to log

node_suffixstr, optional

A suffix to add to the node name for more specific logging

Returns

GraphState

A new state object with the updated trace information

property node_name: str

Get the node name from parameters.