Skip to content

Workflows

The workflow layer orchestrates the pluggable pipeline by composing extractors, gatherers, verifiers, and formatters into a complete evaluation workflow.

WorkflowConfig

truthfulness_evaluator.llm.workflows.config.WorkflowConfig dataclass

Configuration for a truthfulness evaluation workflow.

Bundles ready-to-use strategy instances with workflow-level settings. Callers construct strategy instances with desired config, then hand them to WorkflowConfig. The builder plugs them into graph nodes.

Source code in src/truthfulness_evaluator/llm/workflows/config.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@dataclass
class WorkflowConfig:
    """Configuration for a truthfulness evaluation workflow.

    Bundles ready-to-use strategy instances with workflow-level settings.
    Callers construct strategy instances with desired config, then hand
    them to WorkflowConfig. The builder plugs them into graph nodes.
    """

    name: str
    description: str

    # Strategy instances (must satisfy the relevant protocol)
    extractor: ClaimExtractor
    gatherers: list[EvidenceGatherer]
    verifier: ClaimVerifier
    formatters: list[ReportFormatter]

    # Workflow-level configuration
    max_claims: int | None = None
    enable_human_review: bool = False
    human_review_threshold: float = 0.6

Usage Example:

from truthfulness_evaluator.llm.workflows.config import WorkflowConfig
from truthfulness_evaluator import SimpleExtractor
from truthfulness_evaluator import WebSearchGatherer
from truthfulness_evaluator import ConsensusVerifier
from truthfulness_evaluator import MarkdownFormatter

config = WorkflowConfig(
    extractor=SimpleExtractor(),
    gatherer=WebSearchGatherer(max_results=5),
    verifier=ConsensusVerifier(model_names=["gpt-4o", "claude-3-5-sonnet-20241022"]),
    formatter=MarkdownFormatter()
)

WorkflowRegistry

truthfulness_evaluator.llm.workflows.registry.WorkflowRegistry

Registry for workflow configurations.

Provides: - Built-in presets (external, internal, scientific, etc.) - Runtime registration for custom workflows - Entry-point discovery for third-party plugins

Source code in src/truthfulness_evaluator/llm/workflows/registry.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class WorkflowRegistry:
    """Registry for workflow configurations.

    Provides:
    - Built-in presets (external, internal, scientific, etc.)
    - Runtime registration for custom workflows
    - Entry-point discovery for third-party plugins
    """

    _workflows: dict[str, WorkflowConfig] = {}
    _discovered: bool = False

    @classmethod
    def reset(cls) -> None:
        """Reset registry state. Intended for test isolation."""
        cls._workflows.clear()
        cls._discovered = False

    @classmethod
    def register(
        cls,
        name: str,
        config: WorkflowConfig,
        *,
        override: bool = False,
    ) -> None:
        """Register a workflow configuration.

        Args:
            name: Unique workflow name.
            config: The workflow configuration.
            override: If True, allow overriding existing registrations.

        Raises:
            ValueError: If name is already registered and override is False.
        """
        if name in cls._workflows and not override:
            raise ValueError(f"Workflow '{name}' is already registered")
        cls._workflows[name] = config

    @classmethod
    def get(cls, name: str) -> WorkflowConfig:
        """Get a workflow configuration by name.

        Args:
            name: Workflow name.

        Returns:
            The WorkflowConfig.

        Raises:
            KeyError: If workflow name is not registered.
        """
        cls._discover_plugins()
        if name not in cls._workflows:
            available = ", ".join(sorted(cls._workflows.keys()))
            raise KeyError(f"Unknown workflow '{name}'. Available: {available}")
        return cls._workflows[name]

    @classmethod
    def list_workflows(cls) -> dict[str, str]:
        """List all registered workflows with descriptions.

        Returns:
            Dict of name -> description.
        """
        cls._discover_plugins()
        return {name: config.description for name, config in sorted(cls._workflows.items())}

    @classmethod
    def _discover_plugins(cls) -> None:
        """Discover workflows from entry points (lazy, one-time)."""
        if cls._discovered:
            return
        cls._discovered = True

        try:
            from importlib.metadata import entry_points

            eps = entry_points(group="truthfulness_evaluator.workflows")
            for ep in eps:
                try:
                    config = ep.load()
                    if isinstance(config, WorkflowConfig):
                        cls.register(ep.name, config)
                except Exception as e:
                    logger.warning(f"Failed to load workflow plugin '{ep.name}': {e}")
        except Exception as e:
            logger.debug(f"Entry point discovery not available: {e}")

get(name) classmethod

Get a workflow configuration by name.

Parameters:

Name Type Description Default
name str

Workflow name.

 required

Returns:

Type Description
WorkflowConfig

The WorkflowConfig.

Raises:

Type Description
KeyError

If workflow name is not registered.
Source code in src/truthfulness_evaluator/llm/workflows/registry.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@classmethod
def get(cls, name: str) -> WorkflowConfig:
    """Get a workflow configuration by name.

    Args:
        name: Workflow name.

    Returns:
        The WorkflowConfig.

    Raises:
        KeyError: If workflow name is not registered.
    """
    cls._discover_plugins()
    if name not in cls._workflows:
        available = ", ".join(sorted(cls._workflows.keys()))
        raise KeyError(f"Unknown workflow '{name}'. Available: {available}")
    return cls._workflows[name]

list_workflows() classmethod

List all registered workflows with descriptions.

Returns:

Type Description
dict[str, str]

Dict of name -> description.
Source code in src/truthfulness_evaluator/llm/workflows/registry.py 
68
69
70
71
72
73
74
75
76
@classmethod
def list_workflows(cls) -> dict[str, str]:
    """List all registered workflows with descriptions.

    Returns:
        Dict of name -> description.
    """
    cls._discover_plugins()
    return {name: config.description for name, config in sorted(cls._workflows.items())}

register(name, config, *, override=False) classmethod

Register a workflow configuration.

Parameters:

Name Type Description Default
name str

Unique workflow name.

 required
config WorkflowConfig

The workflow configuration.

 required
override bool

If True, allow overriding existing registrations.

 False

Raises:

Type Description
ValueError

If name is already registered and override is False.
Source code in src/truthfulness_evaluator/llm/workflows/registry.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@classmethod
def register(
    cls,
    name: str,
    config: WorkflowConfig,
    *,
    override: bool = False,
) -> None:
    """Register a workflow configuration.

    Args:
        name: Unique workflow name.
        config: The workflow configuration.
        override: If True, allow overriding existing registrations.

    Raises:
        ValueError: If name is already registered and override is False.
    """
    if name in cls._workflows and not override:
        raise ValueError(f"Workflow '{name}' is already registered")
    cls._workflows[name] = config

reset() classmethod

Reset registry state. Intended for test isolation.

Source code in src/truthfulness_evaluator/llm/workflows/registry.py 
21
22
23
24
25
@classmethod
def reset(cls) -> None:
    """Reset registry state. Intended for test isolation."""
    cls._workflows.clear()
    cls._discovered = False

Usage Example:

from truthfulness_evaluator.llm.workflows.registry import WorkflowRegistry

registry = WorkflowRegistry()

# Get a preset workflow
config = registry.get_workflow("external")

# List all available workflows
workflows = registry.list_workflows()
print(workflows)  # ["external", "full", "quick", "internal"]

WorkflowState

truthfulness_evaluator.llm.workflows.state.WorkflowState

Bases: TypedDict

Unified state for all truthfulness evaluation workflows.

The core fields are used by every workflow. The extensions field provides a namespace for strategy-specific state (e.g., claim classifications for internal verification workflows).

Source code in src/truthfulness_evaluator/llm/workflows/state.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class WorkflowState(TypedDict):
    """Unified state for all truthfulness evaluation workflows.

    The core fields are used by every workflow. The `extensions` field
    provides a namespace for strategy-specific state (e.g., claim
    classifications for internal verification workflows).
    """

    # Input
    document: str
    document_path: str

    # Core pipeline state
    claims: list[Claim]
    current_claim_index: int
    evidence_cache: dict[str, list[Evidence]]
    verifications: list[VerificationResult]

    # Output
    final_report: TruthfulnessReport | None

    # Strategy-specific state (open for extension, closed for modification)
    extensions: dict[str, Any]

Usage Note: WorkflowState is the unified state TypedDict used by LangGraph workflows. It tracks the document, extracted claims, evidence, verification results, and final report throughout the pipeline.

Presets

The workflows.presets module provides pre-configured workflows for common use cases:

Preset Description Extractor Gatherer Verifier Formatter
external Web-based verification with multi-model consensus Simple WebSearch Consensus (3 models) Markdown
full Comprehensive verification with web + filesystem Simple Composite (web + filesystem) Consensus (3 models) Markdown
quick Fast single-model verification Simple WebSearch (limited) Single model Markdown
internal Code-documentation alignment Simple Filesystem Internal (AST/config) Markdown

Usage Example:

from truthfulness_evaluator.llm.workflows.presets import create_external_config

config = create_external_config()

For the internal preset, you must provide the codebase root path:

from truthfulness_evaluator.llm.workflows.presets import create_internal_config

config = create_internal_config(root_path="/path/to/project")