plexus.scores.AgenticValidator module

class plexus.scores.AgenticValidator.AgenticValidator(**parameters)

Bases: LangGraphScore

An agentic validator that uses LangGraph and advanced LangChain components to validate education information, specifically for degree, using both transcript and metadata.

This validator uses a language model to analyze transcripts and validate educational claims through a multi-step workflow implemented with LangGraph.

Initialize the AgenticValidator with the given parameters.

Args:

**parameters: Keyword arguments for configuring the validator.

class Input(*, text: str, metadata: ~typing.Dict[str, ~typing.Any] = <factory>, results: ~typing.List[~typing.Any] | None = None)

Bases: Input

Model input containing the transcript and metadata.

Attributes:

metadata (Dict[str, Any]): A dictionary containing degree information.

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.

classmethod handle_nan(v)

metadata: Dict[str, Any]

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

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

class Parameters(*, scorecard_name: str | None = None, name: str | None = None, id: str | int | None = None, key: str | None = None, dependencies: ~typing.List[dict] | None = None, data: dict | None = None, number_of_classes: int | None = None, label_score_name: str | None = None, label_field: str | None = None, model_provider: ~typing.Literal['ChatOpenAI', 'AzureChatOpenAI', 'BedrockChat', 'ChatVertexAI'] = 'AzureChatOpenAI', model_name: str | None = None, model_region: str | None = None, temperature: float | None = 0, max_tokens: int | None = 500, graph: list[dict] | None = None, input: dict | None = None, output: dict | None = None, depends_on: ~typing.List[str] | ~typing.Dict[str, str | ~typing.Dict[str, ~typing.Any]] | None = None, single_line_messages: bool = False, checkpoint_db_path: str | None = './.plexus/checkpoints/langgraph.db', thread_id: str | None = None, postgres_url: str | None = None, labels: ~typing.List[str] = <factory>, prompt: str = '', dependency: ~typing.Dict[str, str] | None = None, agent_type: ~typing.Literal['react', 'langgraph'] = 'react')

Bases: Parameters

Parameters for configuring the AgenticValidator.

Attributes:

labels (List[str]): The labels of the metadata to validate. prompt (str): The custom prompt to use for validation. dependency (Optional[Dict[str, str]]): The dependency configuration. agent_type (Literal): The type of agent to use for validation.

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.

agent_type: Literal['react', 'langgraph']

dependency: Dict[str, str] | None

labels: List[str]

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

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

prompt: str

class ReActAgentOutputParser(*args: Any, name: str | None = None)

Bases: ReActSingleInputOutputParser

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

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

parse(text: str) → AgentAction | AgentFinish

Parse text into agent action/finish.

__init__(**parameters)

Initialize the AgenticValidator with the given parameters.

Args:

**parameters: Keyword arguments for configuring the validator.

create_lcel_agent()

Create an LCEL-based agent for validation tasks with memory for the transcript.

initialize_validation_workflow()

Initialize the language model and create the workflow.

predict(context, model_input: Input) → Result

Predict the validity of the education information based on the transcript and metadata.

Args:

model_input (LangGraphScore.Input): The input containing the transcript and metadata.

Returns:

LangGraphScore.Result: The output containing the validation result.

class plexus.scores.AgenticValidator.GraphState

Bases: TypedDict

all_information_provided: bool

parsed_schools: List[SchoolInfo]

schools_mentioned: List[str]

text: str

class plexus.scores.AgenticValidator.SchoolInfo(*, school_name: str, modality: str, program: str, level: str)

Bases: BaseModel

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.

level: str

modality: str

model_config: ClassVar[ConfigDict] = {}

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

program: str

school_name: str

class plexus.scores.AgenticValidator.TextAnalysis(*, schools: List[SchoolInfo])

Bases: BaseModel

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] = {}

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

schools: List[SchoolInfo]

class plexus.scores.AgenticValidator.ValidationState(text: str, metadata: ~typing.Dict[str, str], current_step: str = '', validation_result: str = 'Unknown', explanation: str = '', messages: ~typing.List[~langchain_core.messages.human.HumanMessage | ~langchain_core.messages.ai.AIMessage] = <factory>, has_dependency: bool = False)

Bases: object

Represents the state of the validation process at any given point.

This class encapsulates all the information needed to track the progress and results of the validation workflow.

Attributes:

text (str): The text being validated. metadata (Dict[str, str]): Metadata about the education claim being validated. current_step (str): The current step in the validation process. validation_result (str): Result of the degree validation. explanation (str): Detailed explanation of the validation result. messages (List[Union[HumanMessage, AIMessage]]): History of messages in the validation process. has_dependency (bool): Indicates whether there is a dependency prompt.

__init__(text: str, metadata: ~typing.Dict[str, str], current_step: str = '', validation_result: str = 'Unknown', explanation: str = '', messages: ~typing.List[~langchain_core.messages.human.HumanMessage | ~langchain_core.messages.ai.AIMessage] = <factory>, has_dependency: bool = False) → None

current_step: str = ''

explanation: str = ''

has_dependency: bool = False

messages: List[HumanMessage | AIMessage]

metadata: Dict[str, str]

text: str

validation_result: str = 'Unknown'