Skip to content

analyzer

agent_cover.instrumentation.analyzer

Module for analyzing agent outputs against configured decisions.

This module implements the Verification Layer of AgentCover. While other modules track if code ran, this module tracks what the code produced.

Flow
  1. Interception: Instrumentors capture generated text.
  2. Normalization: OutputAnalyzer converts inputs (Strings, Pydantic, Dicts) to a standard dict.
  3. Evaluation: Compares data against DecisionConfig.

Classes

OutputAnalyzer

Analyzes agent outputs to verify business logic coverage.

This class decouples data extraction from validation. It is robust to different data formats, capable of parsing JSON strings embedded in Markdown or handling raw Pydantic models.

Key Features
  • Adapter System: Automatically handles Pydantic V1/V2 and Dataclasses.
  • Fuzzy JSON Parsing: Can extract JSON from messy LLM outputs (e.g., wrapped in markdown code blocks).
  • Decision Matching: Registers hits in the registry when output fields match expected values.

Attributes:

Name Type Description
registry AgentRegistry

Where decision hits are recorded.
config AgentCoverConfig

Source of the decision rules.
adapter_registry AdapterRegistry

Registry for converting objects to dicts.
Source code in src/agent_cover/instrumentation/analyzer.py 
 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
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
class OutputAnalyzer:
    """Analyzes agent outputs to verify business logic coverage.

    This class decouples data extraction from validation. It is robust to
    different data formats, capable of parsing JSON strings embedded in Markdown
    or handling raw Pydantic models.

    Key Features:
        - **Adapter System**: Automatically handles Pydantic V1/V2 and Dataclasses.
        - **Fuzzy JSON Parsing**: Can extract JSON from messy LLM outputs (e.g., wrapped in markdown code blocks).
        - **Decision Matching**: Registers hits in the registry when output fields match expected values.

    Attributes:
        registry (AgentRegistry): Where decision hits are recorded.
        config (AgentCoverConfig): Source of the decision rules.
        adapter_registry (AdapterRegistry): Registry for converting objects to dicts.
    """

    def __init__(
        self,
        registry: Optional[AgentRegistry] = None,
        config: Optional[AgentCoverConfig] = None,
        adapter_registry: Optional[AdapterRegistry] = None,
    ):
        """Initializes the OutputAnalyzer.

        Args:
            registry: The AgentRegistry instance.
            config: The AgentCoverConfig instance.
            adapter_registry: The adapter registry to use for data conversion.
        """
        # Use provided registry or get the default global registry
        self.registry = registry or get_registry()
        # Use provided config or get the default global config
        self.config = config or get_config()
        # Use the injected adapter registry or the default global adapter registry
        self.adapter_registry = adapter_registry or get_default_adapter_registry()

    def analyze(self, payload: Any) -> None:
        r"""Analyzes the given payload to detect matches against configured decisions.

        The method handles three types of payloads:
        1. **Objects**: Uses adapters to convert Pydantic/Dataclasses to dicts.
        2. **Dictionaries**: Analyzes fields directly.
        3. **Strings**: Attempts to parse JSON. If parsing fails, performs substring matching.

        Args:
            payload (Any): The output to analyze. Can be a Dict, str, Pydantic Model, etc.

        Examples:
            >>> analyzer.analyze({"intent": "REFUND", "confidence": 0.9})
            # Registers a hit for decision with target_field="intent" and expected_values=["REFUND"]

            >>> analyzer.analyze("The user wants to buy something.")
            # Registers a hit if a decision expects "buy" in the raw text.

            >>> analyzer.analyze("```json\n{'status': 'DONE'}\n```")
            # fuzzy parses JSON and matches status=DONE.
        """
        if not payload:
            return

        # 1. Normalization
        data = None

        # Retrieve the specific adapter for this payload instance
        adapter = self.adapter_registry.get_adapter_for_instance(payload)

        if adapter:
            try:
                # Try to convert the payload to a dictionary using the adapter
                data = adapter.to_dict(payload)
            except Exception as e:
                logger.warning(e, exc_info=True)
        elif isinstance(payload, dict):
            # If payload is already a dictionary, use it directly
            data = payload
        elif isinstance(payload, str):
            # Attempt to parse a JSON string
            try:
                clean_str = payload.strip()

                # Handle markdown code blocks containing JSON
                if "```json" in clean_str:
                    match = re.search(r"```json(.*?)```", clean_str, re.DOTALL)
                    if match:
                        clean_str = match.group(1).strip()
                elif "```" in clean_str:
                    match = re.search(r"```(.*?)```", clean_str, re.DOTALL)
                    if match:
                        clean_str = match.group(1).strip()

                # If the string looks like a JSON object, attempt to load it
                if clean_str.startswith("{"):
                    data = json.loads(clean_str)
            except Exception as e:
                logger.warning(e, exc_info=True)

        # 2. Matching Logic
        if not self.config.decisions:
            return

        # Case A: Payload is a dictionary (structured data)
        if isinstance(data, dict):
            for decision in self.config.decisions:
                # Check if the target field exists in the data
                if decision.target_field in data:
                    val = data[decision.target_field]
                    # Convert the value to string (handling Enums if present)
                    val_str = val.name if hasattr(val, "name") else str(val)

                    # Check if the value matches any of the expected values
                    if val_str in decision.expected_values:
                        self.registry.register_decision_hit(decision.id, val_str)

        # Case B: Payload is a string (unstructured text)
        elif isinstance(payload, str):
            clean_text = payload.strip()
            for decision in self.config.decisions:
                for expected in decision.expected_values:
                    # Check if the expected value is present in the text
                    if expected in clean_text:
                        self.registry.register_decision_hit(decision.id, expected)
Functions
__init__(registry=None, config=None, adapter_registry=None)

Initializes the OutputAnalyzer.

Parameters:

Name Type Description Default
registry Optional[AgentRegistry]

The AgentRegistry instance.

 None
config Optional[AgentCoverConfig]

The AgentCoverConfig instance.

 None
adapter_registry Optional[AdapterRegistry]

The adapter registry to use for data conversion.

 None
Source code in src/agent_cover/instrumentation/analyzer.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    registry: Optional[AgentRegistry] = None,
    config: Optional[AgentCoverConfig] = None,
    adapter_registry: Optional[AdapterRegistry] = None,
):
    """Initializes the OutputAnalyzer.

    Args:
        registry: The AgentRegistry instance.
        config: The AgentCoverConfig instance.
        adapter_registry: The adapter registry to use for data conversion.
    """
    # Use provided registry or get the default global registry
    self.registry = registry or get_registry()
    # Use provided config or get the default global config
    self.config = config or get_config()
    # Use the injected adapter registry or the default global adapter registry
    self.adapter_registry = adapter_registry or get_default_adapter_registry()
analyze(payload)

Analyzes the given payload to detect matches against configured decisions.

The method handles three types of payloads: 1. Objects: Uses adapters to convert Pydantic/Dataclasses to dicts. 2. Dictionaries: Analyzes fields directly. 3. Strings: Attempts to parse JSON. If parsing fails, performs substring matching.

Parameters:

Name Type Description Default
payload Any

The output to analyze. Can be a Dict, str, Pydantic Model, etc.

 required

Examples:

>>> analyzer.analyze({"intent": "REFUND", "confidence": 0.9})
# Registers a hit for decision with target_field="intent" and expected_values=["REFUND"]

>>> analyzer.analyze("The user wants to buy something.")
# Registers a hit if a decision expects "buy" in the raw text.

>>> analyzer.analyze("```json\n{'status': 'DONE'}\n```")
# fuzzy parses JSON and matches status=DONE.
Source code in src/agent_cover/instrumentation/analyzer.py 
 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
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def analyze(self, payload: Any) -> None:
    r"""Analyzes the given payload to detect matches against configured decisions.

    The method handles three types of payloads:
    1. **Objects**: Uses adapters to convert Pydantic/Dataclasses to dicts.
    2. **Dictionaries**: Analyzes fields directly.
    3. **Strings**: Attempts to parse JSON. If parsing fails, performs substring matching.

    Args:
        payload (Any): The output to analyze. Can be a Dict, str, Pydantic Model, etc.

    Examples:
        >>> analyzer.analyze({"intent": "REFUND", "confidence": 0.9})
        # Registers a hit for decision with target_field="intent" and expected_values=["REFUND"]

        >>> analyzer.analyze("The user wants to buy something.")
        # Registers a hit if a decision expects "buy" in the raw text.

        >>> analyzer.analyze("```json\n{'status': 'DONE'}\n```")
        # fuzzy parses JSON and matches status=DONE.
    """
    if not payload:
        return

    # 1. Normalization
    data = None

    # Retrieve the specific adapter for this payload instance
    adapter = self.adapter_registry.get_adapter_for_instance(payload)

    if adapter:
        try:
            # Try to convert the payload to a dictionary using the adapter
            data = adapter.to_dict(payload)
        except Exception as e:
            logger.warning(e, exc_info=True)
    elif isinstance(payload, dict):
        # If payload is already a dictionary, use it directly
        data = payload
    elif isinstance(payload, str):
        # Attempt to parse a JSON string
        try:
            clean_str = payload.strip()

            # Handle markdown code blocks containing JSON
            if "```json" in clean_str:
                match = re.search(r"```json(.*?)```", clean_str, re.DOTALL)
                if match:
                    clean_str = match.group(1).strip()
            elif "```" in clean_str:
                match = re.search(r"```(.*?)```", clean_str, re.DOTALL)
                if match:
                    clean_str = match.group(1).strip()

            # If the string looks like a JSON object, attempt to load it
            if clean_str.startswith("{"):
                data = json.loads(clean_str)
        except Exception as e:
            logger.warning(e, exc_info=True)

    # 2. Matching Logic
    if not self.config.decisions:
        return

    # Case A: Payload is a dictionary (structured data)
    if isinstance(data, dict):
        for decision in self.config.decisions:
            # Check if the target field exists in the data
            if decision.target_field in data:
                val = data[decision.target_field]
                # Convert the value to string (handling Enums if present)
                val_str = val.name if hasattr(val, "name") else str(val)

                # Check if the value matches any of the expected values
                if val_str in decision.expected_values:
                    self.registry.register_decision_hit(decision.id, val_str)

    # Case B: Payload is a string (unstructured text)
    elif isinstance(payload, str):
        clean_text = payload.strip()
        for decision in self.config.decisions:
            for expected in decision.expected_values:
                # Check if the expected value is present in the text
                if expected in clean_text:
                    self.registry.register_decision_hit(decision.id, expected)

Functions