API Reference

Protocols

FeedbackProtocol

source

Protocol for submitting and querying AI feedback.

submit_feedback

async def submit_feedback(
    trace_id: str,
    score: float,
    comment: str | None = None,
    metadata: dict[str, Any] | None = None
) -> None

source

Submit feedback for an AI generation.

get_feedback_stats

async def get_feedback_stats(
    model: str | None = None,
    provider: str | None = None
) -> dict[str, Any]

source

Query aggregate feedback statistics.

---

FeedbackStoreProtocol

source

Persist and query collected feedback items.

All implementations must be async and treat save() as safe to call from request-handling hot-paths (synchronous returns acceptable for cache misses).

save

async def save(feedback: FeedbackItem) -> Result[str, Exception]

source

Persist a single feedback item.

Parameters

Parameter	Type	Description
`feedback`	FeedbackItem	The feedback item to store.

Returns

Type	Description
Result[str, Exception]	Ok(feedback.id) on success, Err(exception) on failure.

find_by_session

async def find_by_session(session_id: str) -> list[FeedbackItem]

source

Retrieve all feedback items for a session.

Parameters

Parameter	Type	Description
`session_id`	str	Session identifier from feedback context.

Returns

Type	Description
list[FeedbackItem]	All collected items in that session, newest first.

find_by_type

async def find_by_type(
    feedback_type: FeedbackType,
    *,
    limit: int = 100
) -> list[FeedbackItem]

source

Retrieve feedback items of a given type.

Parameters

Parameter	Type	Description
`feedback_type`	FeedbackType	The type to filter by.
`limit`	int	Maximum number of results (default 100).

Returns

Type	Description
list[FeedbackItem]	Matching items ordered by creation time descending.

aggregate

async def aggregate(
    *,
    window_hours: int = 24
) -> FeedbackSummary

source

Compute summary statistics for items in a time window.

Parameters

Parameter	Type	Description
`window_hours`	int	Look-back window in hours (default 24).

Returns

Type	Description
FeedbackSummary	Aggregated FeedbackSummary statistics.

---

Classes

CachedFeedbackStore

source

Write-through cache: cold reads from *store*, hot reads from *cache*.

Session-scoped and type-scoped result lists are cached with short TTLs. Any save call invalidates the relevant cache entries so subsequent reads always reflect the latest data.

Parameters

Parameter	Type	Description
`store`		Durable backing store (e.g. DatabaseFeedbackStore).
`cache`		Cache backend used for hot reads.

__init__

def __init__(
    store: FeedbackStoreProtocol,
    cache: CacheBackendProtocol
) -> None

source

save

async def save(feedback: FeedbackItem) -> Result[str, Exception]

source

Persist feedback and invalidate related cache entries.

Parameters

Parameter	Type	Description
`feedback`	FeedbackItem	Item to persist.

Returns

Type	Description
Result[str, Exception]	Result from the backing store.

find_by_session

async def find_by_session(session_id: str) -> list[FeedbackItem]

source

Return items for session_id, using cache when available.

Parameters

Parameter	Type	Description
`session_id`	str	Session identifier.

Returns

Type	Description
list[FeedbackItem]	Feedback items for the session, newest first.

find_by_type

async def find_by_type(
    feedback_type: FeedbackType,
    *,
    limit: int = 100
) -> list[FeedbackItem]

source

Return items of feedback_type, using cache when available.

Parameters

Parameter	Type	Description
`feedback_type`	FeedbackType	Type to filter by.
`limit`	int	Maximum result count.

Returns

Type	Description
list[FeedbackItem]	Matching feedback items, newest first.

aggregate

async def aggregate(
    *,
    window_hours: int = 24
) -> FeedbackSummary

source

Delegate aggregation to the backing store — not cached.

Parameters

Parameter	Type	Description
`window_hours`	int	Look-back window in hours.

Returns

Type	Description
FeedbackSummary	FeedbackSummary.

---

DatabaseFeedbackStore

source

Persist and query feedback items via DatabaseProviderProtocol.

The backing table (ai_feedback) is created lazily on first write. Indexed columns: session_id, type, created_at.

Parameters

Parameter	Type	Description
`provider`		DI-injected database provider.

__init__

def __init__(provider: DatabaseProviderProtocol) -> None

source

save

async def save(feedback: FeedbackItem) -> Result[str, FeedbackError]

source

Persist feedback and return its ID on success.

Parameters

Parameter	Type	Description
`feedback`	FeedbackItem	Item to store.

Returns

Type	Description
Result[str, FeedbackError]	``Ok(feedback.id)`` on success, ``Err(FeedbackError)`` on failure.

find_by_session

async def find_by_session(session_id: str) -> list[FeedbackItem]

source

Return all items collected during session_id, newest first.

Parameters

Parameter	Type	Description
`session_id`	str	Session identifier to filter by.

Returns

Type	Description
list[FeedbackItem]	Matching feedback items ordered by creation time descending.

find_by_type

async def find_by_type(
    feedback_type: FeedbackType,
    *,
    limit: int = 100
) -> list[FeedbackItem]

source

Return feedback items of feedback_type, newest first.

Parameters

Parameter	Type	Description
`feedback_type`	FeedbackType	Type to filter by.
`limit`	int	Maximum number of results (default: 100).

Returns

Type	Description
list[FeedbackItem]	Matching feedback items ordered by creation time descending.

aggregate

async def aggregate(
    *,
    window_hours: int = 24
) -> FeedbackSummary

source

Compute summary statistics for items in the last window_hours hours.

Parameters

Parameter	Type	Description
`window_hours`	int	Look-back window in hours (default: 24).

Returns

Type	Description
FeedbackSummary	Aggregated FeedbackSummary for the window.

---

FeedbackCollector

source

Collects and stores user feedback.

Provides methods to capture different types of feedback and integrate with ML retraining pipelines.

Example

collector = FeedbackCollector()
# Collect rating
await collector.collect_rating(
rating=5,
context={"model": "gpt-4", "input": "Hello"}
)
# Collect correction
await collector.collect_correction(
original="incorrect output",
corrected="correct output",
context={"model_id": "123"}
)
# Get all feedback
items = collector.get_feedback()

__init__

def __init__(storage: FeedbackStoreProtocol | None = None)

source

Initialize feedback collector.

Parameters

Parameter	Type	Description
`storage`	FeedbackStoreProtocol | None	Optional storage backend (e.g., database, file) If None, uses in-memory storage

collect_rating

async def collect_rating(
    rating: float,
    context: dict[str, Any] | None = None,
    metadata: dict[str, Any] | None = None
) -> str

source

Collect a rating feedback.

Parameters

Parameter	Type	Description
`rating`	float	Numeric rating value
`context`	dict[str, Any] | None	Context about what was rated
`metadata`	dict[str, Any] | None	Additional metadata

Returns

Type	Description
str	Feedback ID

collect_text

async def collect_text(
    text: str,
    context: dict[str, Any] | None = None,
    metadata: dict[str, Any] | None = None
) -> str

source

Collect text feedback.

Parameters

Parameter	Type	Description
`text`	str	Feedback text
`context`	dict[str, Any] | None	Context about what feedback is for
`metadata`	dict[str, Any] | None	Additional metadata

Returns

Type	Description
str	Feedback ID

collect_correction

async def collect_correction(
    original: str,
    corrected: str,
    context: dict[str, Any] | None = None,
    metadata: dict[str, Any] | None = None
) -> str

source

Collect a correction feedback.

Parameters

Parameter	Type	Description
`original`	str	Original (incorrect) output
`corrected`	str	Corrected output
`context`	dict[str, Any] | None	Context
`metadata`	dict[str, Any] | None	Additional metadata

Returns

Type	Description
str	Feedback ID

collect_label

async def collect_label(
    label: Any,
    input_data: Any,
    context: dict[str, Any] | None = None,
    metadata: dict[str, Any] | None = None
) -> str

source

Collect ground truth label.

Useful for supervised learning and model retraining.

Parameters

Parameter	Type	Description
`label`	Any	Ground truth label
`input_data`	Any	Input that this label corresponds to
`context`	dict[str, Any] | None	Context
`metadata`	dict[str, Any] | None	Additional metadata

Returns

Type	Description
str	Feedback ID

get_feedback

async def get_feedback(
    feedback_type: FeedbackType | None = None,
    limit: int | None = None
) -> list[FeedbackItem]

source

Get feedback items.

Parameters

Parameter	Type	Description
`feedback_type`	FeedbackType | None	Optional filter by type
`limit`	int | None	Maximum number of items to return

Returns

Type	Description
list[FeedbackItem]	List of feedback items from memory or the storage backend.

get_feedback_dict

async def get_feedback_dict(
    feedback_type: FeedbackType | None = None,
    limit: int | None = None
) -> list[dict[str, Any]]

source

Get feedback as dictionaries.

Parameters

Parameter	Type	Description
`feedback_type`	FeedbackType | None	Optional filter by type
`limit`	int | None	Maximum number of items

Returns

Type	Description
list[dict[str, Any]]	List of feedback dictionaries

clear

def clear() -> None

source

Clear all feedback from the in-memory buffer.

---

FeedbackConfig

source

Configuration for AI feedback collection.

Loaded from the ai_feedback: key in application.yaml, with environment variable overrides via LEX_AI_FEEDBACK__* prefix.

validate_for_environment

def validate_for_environment(env: Environment | None = None) -> list[ConfigIssue]

source

Check config is safe for the target environment.

---

FeedbackContext

source

Context manager for feedback collection during operations.

Automatically captures context and results for feedback.

Example

collector = FeedbackCollector()
async with FeedbackContext(collector, operation="prediction") as ctx:
result = await model.predict(input_data)
ctx.set_result(result)
# Feedback context is automatically stored

__init__

def __init__(
    collector: FeedbackCollector,
    operation: str,
    metadata: dict[str, Any] | None = None
)

source

Initialize feedback context.

Parameters

Parameter	Type	Description
`collector`	FeedbackCollector	Feedback collector
`operation`	str	Operation name
`metadata`	dict[str, Any] | None	Additional metadata

set_input

def set_input(input_data: Any) -> None

source

Set input data.

Parameters

Parameter	Type	Description
`input_data`	Any	Input to the operation

set_result

def set_result(result: Any) -> None

source

Set operation result.

Parameters

Parameter	Type	Description
`result`	Any	Result of the operation

---

FeedbackItem

source

A single feedback item.

Attributes: feedback_type: Type of feedback (rating, text, correction, or label). value: The feedback value (e.g. rating score, text comment). context: Context about what was being evaluated (session_id, model, etc.). metadata: Additional metadata dictionary. id: Unique feedback identifier. created_at: Timestamp when feedback was created.

type

property type() -> FeedbackType

source

Alias for feedback_type for backward compatibility.

Returns

Type	Description
FeedbackType	The feedback type.

to_dict

def to_dict() -> dict[str, Any]

source

Convert to dictionary.

Returns

Type	Description
dict[str, Any]	Dictionary representation with ISO-formatted timestamp.

---

FeedbackMiddleware

source

Middleware for automatic feedback capture.

Captures prediction inputs/outputs automatically and provides hooks for user feedback collection.

Example

from lexigram.app import Application
from lexigram.ai.feedback import FeedbackMiddleware, FeedbackCollector

app = Application()
collector = FeedbackCollector()
middleware = FeedbackMiddleware(collector)
app.use(middleware)

__init__

def __init__(
    collector: FeedbackCollector,
    capture_inputs: bool = True,
    capture_outputs: bool = True,
    capture_metadata: bool = True,
    registry: FeedbackProcessorRegistry | None = None
)

source

Initialize feedback middleware.

Parameters

Parameter	Type	Description
`collector`	FeedbackCollector	Feedback collector instance
`capture_inputs`	bool	Whether to capture request inputs
`capture_outputs`	bool	Whether to capture response outputs
`capture_metadata`	bool	Whether to capture additional metadata
`registry`	FeedbackProcessorRegistry | None	Feedback processor registry (defaults to one with built-in processors)

create_feedback_endpoint

def create_feedback_endpoint() -> Callable

source

Create a feedback submission endpoint.

Returns

Type	Description
Callable	Async handler function for feedback submission

Example

middleware = FeedbackMiddleware(collector)
feedback_handler = middleware.create_feedback_endpoint()
# Use with your web framework:
# app.post("/feedback", feedback_handler)

---

FeedbackModule

source

AI Feedback collection and processing integration.

Call configure to register a FeedbackProtocol implementation along with the processor registry and storage backend for injection.

Usage

from lexigram.ai.feedback.config import FeedbackConfig

@module(
    imports=[
        FeedbackModule.configure(
            FeedbackConfig(storage="database")
        )
    ]
)
class AppModule(Module):
    pass

from lexigram.ai.feedback.config import FeedbackConfig

@module(
    imports=[
        FeedbackModule.configure(
            FeedbackConfig(storage="database")
        )
    ]
)
class AppModule(Module):
    pass

Error Handling

The feedback services use the Result pattern for expected failures.
Domain errors are available via the exported exception hierarchy::

    from lexigram.ai.feedback.exceptions import (
        FeedbackError,           # base — catch-all
        FeedbackProcessingError, # processor pipeline failure
        FeedbackValidationError, # schema / data-validation failure
    )

The feedback services use the Result pattern for expected failures.
Domain errors are available via the exported exception hierarchy

from lexigram.ai.feedback.exceptions import (
    FeedbackError,           # base — catch-all
    FeedbackProcessingError, # processor pipeline failure
    FeedbackValidationError, # schema / data-validation failure
)

    from lexigram.ai.feedback.exceptions import (
        FeedbackError,           # base — catch-all
        FeedbackProcessingError, # processor pipeline failure
        FeedbackValidationError, # schema / data-validation failure
    )

Exports: FeedbackProtocol, FeedbackError, FeedbackProcessingError, FeedbackValidationError

configure

def configure(
    cls,
    config: FeedbackConfig | None = None
) -> DynamicModule

source

Create a FeedbackModule with the given configuration.

Parameters

Parameter	Type	Description
`config`	FeedbackConfig | None	FeedbackConfig, a plain ``dict`` of the same keys, or ``None`` to read from environment variables.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

stub

def stub(
    cls,
    config: FeedbackConfig | None = None
) -> DynamicModule

source

Create a FeedbackModule suitable for unit and integration testing.

Uses in-memory or no-op implementations with minimal side effects.

Parameters

Parameter	Type	Description
`config`	FeedbackConfig | None	Optional config override. Uses safe test defaults when None.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

---

FeedbackProcessedHook

source

Payload fired when feedback processing finishes.

---

FeedbackProvider

source

Provider for AI Feedback.

Registers FeedbackCollector and FeedbackProcessorRegistry. Wires a durable storage backend (DB-backed, optionally cached) into the collector during boot.

__init__

def __init__(config: FeedbackConfig | dict | None = None) -> None

source

from_config

def from_config(
    cls,
    config: FeedbackConfig,
    **context: object
) -> FeedbackProvider

source

Factory method for DI container setup.

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register the feedback services.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Wire a durable storage backend into FeedbackCollector.

shutdown

async def shutdown() -> None

source

Shutdown phase.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Health check — always healthy (in-process domain provider).

No external backend to ping.

Parameters

Parameter	Type	Description
`timeout`	float	Ignored for in-process providers.

Returns

Type	Description
HealthCheckResult	Always HEALTHY — no external backend to ping.

---

FeedbackService

source

High-level feedback submission and querying service.

Implements FeedbackProtocol. Wraps a FeedbackStoreProtocol with trace-id correlation and aggregation logic.

Parameters

Parameter	Type	Description
`store`		Optional storage backend; when ``None`` the service operates in a degraded no-op mode and logs warnings instead of persisting.

__init__

def __init__(store: FeedbackStoreProtocol | None = None) -> None

source

submit_feedback

async def submit_feedback(
    trace_id: str,
    score: float,
    comment: str | None = None,
    metadata: dict[str, Any] | None = None
) -> None

source

Submit feedback for a traced AI generation.

Parameters

Parameter	Type	Description
`trace_id`	str	Identifier of the AI generation trace.
`score`	float	Numeric feedback score (e.g. 0.0–1.0 or 1–5).
`comment`	str | None	Optional free-text comment stored in metadata.
`metadata`	dict[str, Any] | None	Optional additional key-value metadata.

get_feedback_stats

async def get_feedback_stats(
    model: str | None = None,
    provider: str | None = None
) -> dict[str, Any]

source

Get aggregate feedback statistics.

Parameters

Parameter	Type	Description
`model`	str | None	Optional model name for context (currently informational).
`provider`	str | None	Optional provider name for context (currently informational).

Returns

Type	Description
dict[str, Any]	Dictionary with ``total_count``, ``average_rating``, and ``by_type`` breakdown, plus optional ``model``/``provider`` keys when supplied.

---

FeedbackStoredHook

source

Payload fired when feedback is persisted by a feedback store.

---

FeedbackSubmittedEvent

source

Emitted when user feedback is submitted and persisted.

Consumed by: quality tracking, model improvement, audit.

---

FeedbackSubmittedHook

source

Payload fired when feedback is submitted to the feedback pipeline.

---

FeedbackSummary

source

Aggregated statistics for collected feedback items.

Attributes: total_count: Total number of feedback items in the window. average_rating: Mean rating across all RATING-type items, or None if no ratings were collected. count_by_type: Item count keyed by FeedbackType enum value (e.g. “rating”, “text”, “correction”, “label”).

---

FeedbackType

source

Type of feedback collected.

---

Exceptions

FeedbackError

source

Base exception for all feedback-related errors.

---

FeedbackProcessingError

source

Raised when a feedback processor fails.

---

FeedbackValidationError

source

Raised when feedback data fails validation.

---