[PRE-446] Add named quality schema support

README.md

@@ -18,23 +18,33 @@ from langchain.agents import create_agent
 from langchain_core.tools import tool
 from prefactor_langchain import PrefactorMiddleware
 
-_OPS = {ast.Add: operator.add, ast.Sub: operator.sub,
-        ast.Mult: operator.mul, ast.Div: operator.truediv}
+_OPS = {
+    ast.Add: operator.add,
+    ast.Sub: operator.sub,
+    ast.Mult: operator.mul,
+    ast.Div: operator.truediv,
+}
+
 
 def _safe_eval(node):
-    if isinstance(node, ast.Constant): return node.n
-    if isinstance(node, ast.BinOp): return _OPS[type(node.op)](_safe_eval(node.left), _safe_eval(node.right))
-    if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub): return -_safe_eval(node.operand)
+    if isinstance(node, ast.Constant):
+        return node.n
+    if isinstance(node, ast.BinOp):
+        return _OPS[type(node.op)](_safe_eval(node.left), _safe_eval(node.right))
+    if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
+        return -_safe_eval(node.operand)
     raise ValueError(f"Unsupported: {node}")
 
+
 @tool
 def calculator(expression: str) -> str:
     """Evaluate a mathematical expression safely."""
     try:
-        return str(_safe_eval(ast.parse(expression, mode='eval').body))
+        return str(_safe_eval(ast.parse(expression, mode="eval").body))
     except Exception as e:
         return f"Error: {e}"
 
+
 async def main():
     middleware = PrefactorMiddleware.from_config(
         api_url="https://api.prefactor.ai",
@@ -51,10 +61,13 @@ async def main():
 
     # All LLM calls and tool executions are automatically traced
     try:
-        result = await agent.ainvoke({"messages": [{"role": "user", "content": "What is 6 * 7?"}]})
+        result = await agent.ainvoke(
+            {"messages": [{"role": "user", "content": "What is 6 * 7?"}]}
+        )
     finally:
         await middleware.close()
 
+
 asyncio.run(main())
 ```
 

packages/core/README.md

@@ -43,6 +43,7 @@ registry.register_type(
     template="{{model}}: {{prompt}} → {{response}}",
 )
 
+
 async def main():
     config = PrefactorCoreConfig(
         http_config=HttpClientConfig(
@@ -67,6 +68,7 @@ async def main():
 
         await instance.finish()
 
+
 asyncio.run(main())
 ```
 
@@ -105,7 +107,7 @@ async with client.span(
     instance_id="instance_123",
     schema_name="agent:llm",
     parent_span_id=None,  # Optional: auto-detected from context stack if omitted
-    payload=None,         # Optional: used as params if span.start() is never called explicitly
+    payload=None,  # Optional: used as params if span.start() is never called explicitly
 ) as span:
     await span.start({"model": "gpt-4", "prompt": "Hello"})
     result = await call_llm()
@@ -183,9 +185,9 @@ config = PrefactorCoreConfig(
         api_token="your-token",
     ),
     queue_config=QueueConfig(
-        num_workers=3,        # Number of background workers
-        max_retries=3,        # Retries per operation
-        retry_delay_base=1.0, # Base delay (seconds) for exponential backoff
+        num_workers=3,  # Number of background workers
+        max_retries=3,  # Retries per operation
+        retry_delay_base=1.0,  # Base delay (seconds) for exponential backoff
     ),
     schema_registry=None,  # Optional: SchemaRegistry instance
 )

packages/core/src/prefactor_core/client.py

@@ -333,10 +333,12 @@ async def _process_operation(self, operation: Operation) -> None:
                         return
                     raise
 
-            elif operation.type == OperationType.UPDATE_AGENT_INSTANCE:
-                await self._http.agent_instances.update(
+            elif operation.type == OperationType.RECORD_QUALITY:
+                await self._http.agent_instances.record_quality(
                     agent_instance_id=operation.payload["instance_id"],
-                    quality_payload=operation.payload.get("quality_payload"),
+                    name=operation.payload["name"],
+                    payload=operation.payload.get("payload"),
+                    idempotency_key=operation.payload.get("idempotency_key"),
                 )
             elif operation.type == OperationType.CREATE_SPAN:
                 await self._http.agent_spans.create(
@@ -541,23 +543,28 @@ async def finish_span(
             timestamp=timestamp,
         )
 
-    async def update_agent_instance(
+    async def record_quality(
         self,
         instance_id: str,
-        quality_payload: dict[str, Any] | None = None,
+        name: str,
+        payload: dict[str, Any] | None = None,
     ) -> None:
-        """Update an agent instance (e.g., set quality payload).
+        """Record a quality payload on an agent instance.
 
         Args:
             instance_id: The ID of the instance to update.
-            quality_payload: Quality evaluation payload (None to clear).
+            name: Quality schema name (key in the agent schema version
+                quality_schemas).
+            payload: Quality payload for this name, or None to remove the
+                recorded payload for this name.
         """
         self._ensure_initialized()
         assert self._instance_manager is not None
 
-        await self._instance_manager.update(
+        await self._instance_manager.record_quality(
             instance_id,
-            quality_payload=quality_payload,
+            name=name,
+            payload=payload,
         )
 
     @asynccontextmanager

packages/core/src/prefactor_core/managers/agent_instance.py

@@ -189,24 +189,29 @@ async def finish_with_idempotency_key(
 
         await self._enqueue(operation)
 
-    async def update(
+    async def record_quality(
         self,
         instance_id: str,
-        quality_payload: dict[str, Any] | None = None,
+        name: str,
+        payload: dict[str, Any] | None = None,
     ) -> None:
-        """Update an agent instance (e.g., set quality payload).
+        """Record a quality payload on an agent instance.
 
-        Queues an update operation for the instance.
+        Queues a record_quality operation for the instance.
 
         Args:
             instance_id: The ID of the instance to update.
-            quality_payload: Quality evaluation payload (None to clear).
+            name: Quality schema name (key in the agent schema version
+                quality_schemas).
+            payload: Quality payload for this name (None to remove).
         """
         operation = Operation(
-            type=OperationType.UPDATE_AGENT_INSTANCE,
+            type=OperationType.RECORD_QUALITY,
             payload={
                 "instance_id": instance_id,
-                "quality_payload": quality_payload,
+                "name": name,
+                "payload": payload,
+                "idempotency_key": generate_idempotency_key(),
             },
             timestamp=datetime.now(timezone.utc),
         )
@@ -303,22 +308,26 @@ async def finish(
             timestamp=timestamp,
         )
 
-    async def update(
+    async def record_quality(
         self,
-        quality_payload: dict[str, Any] | None = None,
+        name: str,
+        payload: dict[str, Any] | None = None,
     ) -> None:
-        """Update the instance (e.g., set quality payload).
+        """Record a quality payload on the instance.
 
-        This queues an update operation for the instance.
+        This queues a record_quality operation for the instance.
 
         Args:
-            quality_payload: Quality evaluation payload (None to clear).
+            name: Quality schema name (key in the agent schema version
+                quality_schemas).
+            payload: Quality payload for this name (None to remove).
         """
         manager = self._client.instance_manager
         assert manager is not None
-        await manager.update(
+        await manager.record_quality(
             self._instance_id,
-            quality_payload=quality_payload,
+            name=name,
+            payload=payload,
         )
 
     async def create_span(

packages/core/src/prefactor_core/models.py

@@ -33,7 +33,7 @@ class AgentInstance:
     started_at: datetime | None = None
     finished_at: datetime | None = None
     metadata: dict[str, Any] = field(default_factory=dict)
-    quality_payload: dict[str, Any] | None = None
+    quality_payloads: dict[str, dict[str, Any]] | None = None
 
 
 @dataclass

packages/core/src/prefactor_core/operations.py

@@ -19,7 +19,7 @@ class OperationType(Enum):
     REGISTER_AGENT_INSTANCE = auto()
     START_AGENT_INSTANCE = auto()
     FINISH_AGENT_INSTANCE = auto()
-    UPDATE_AGENT_INSTANCE = auto()
+    RECORD_QUALITY = auto()
     CREATE_SPAN = auto()
     FINISH_SPAN = auto()
 

packages/core/src/prefactor_core/schema_registry.py

@@ -25,8 +25,9 @@ class SchemaRegistry:
       description, and template per span type
 
     Use ``register()`` for simple payload schemas, ``register_result()`` to add
-    a result schema for an existing entry, or ``register_type()`` for the full
-    structured form. All three approaches can be mixed; ``to_agent_schema_version()``
+    a result schema for an existing entry, ``register_type()`` for the full
+    structured form, or ``register_quality_schema()`` for named quality
+    schemas. All approaches can be mixed; ``to_agent_schema_version()``
     emits whichever fields are populated.
 
     Example:
@@ -67,8 +68,8 @@ def __init__(self) -> None:
         self._span_result_schemas: dict[str, dict[str, Any]] = {}
         # span_type_schemas: name → full structured entry
         self._span_type_schemas: dict[str, dict[str, Any]] = {}
-        # quality_schema: optional quality schema for instance evaluations
-        self._quality_schema: dict[str, Any] | None = None
+        # quality_schemas: list of named quality schema entries
+        self._quality_schemas: list[dict[str, Any]] = []
 
     def register(
         self,
@@ -192,33 +193,36 @@ def register_type(
 
     def register_quality_schema(
         self,
+        name: str,
         schema: dict[str, Any],
         title: str | None = None,
         description: str | None = None,
         template: str | None = None,
         data_risk: dict[str, Any] | None = None,
     ) -> None:
-        """Register a quality schema for instance evaluations.
+        """Register a named quality schema for instance evaluations.
 
-        The quality schema defines the shape of quality payloads that can be
-        set on agent instances. It uses the same ``title``, ``description``,
-        ``template``, and ``data_risk`` fields as span type schemas.
+        Quality schemas define the shape of quality payloads that can be
+        recorded on agent instances. Multiple quality schemas can be
+        registered, each identified by a unique ``name``.
 
         Args:
+            name: Schema name (key used when recording quality payloads).
             schema: JSON Schema dict defining the quality payload structure.
-            title: Optional human-readable title (defaults to "quality" on API).
+            title: Optional human-readable title (defaults to name on API).
             description: Optional description of the quality evaluation.
             template: Optional display template using ``{{field}}`` interpolation.
             data_risk: Optional data risk classification dict (same structure
                 as span type data_risk).
 
         Raises:
-            ValueError: If a quality schema is already registered.
+            ValueError: If a quality schema with the same name is already
+                registered.
         """
-        if self._quality_schema is not None:
-            raise ValueError("Quality schema is already registered")
+        if any(q["name"] == name for q in self._quality_schemas):
+            raise ValueError(f"Quality schema '{name}' is already registered")
 
-        entry: dict[str, Any] = {"schema": schema}
+        entry: dict[str, Any] = {"name": name, "schema": schema}
         if title is not None:
             entry["title"] = title
         if description is not None:
@@ -228,7 +232,7 @@ def register_quality_schema(
         if data_risk is not None:
             entry["data_risk"] = data_risk
 
-        self._quality_schema = entry
+        self._quality_schemas.append(entry)
 
     def get(self, schema_name: str) -> dict[str, Any] | None:
         """Get a params schema by name.
@@ -265,8 +269,8 @@ def has_schema(self, schema_name: str) -> bool:
     def to_agent_schema_version(self, external_id: str) -> dict[str, Any]:
         """Convert registry contents to API-compatible agent_schema_version format.
 
-        Emits ``span_schemas``, ``span_result_schemas``, and ``span_type_schemas``
-        for whichever have been populated.
+        Emits ``span_schemas``, ``span_result_schemas``, ``span_type_schemas``,
+        and ``quality_schemas`` for whichever have been populated.
 
         Args:
             external_id: External identifier for this combined schema version
@@ -286,8 +290,8 @@ def to_agent_schema_version(self, external_id: str) -> dict[str, Any]:
         if self._span_type_schemas:
             result["span_type_schemas"] = list(self._span_type_schemas.values())
 
-        if self._quality_schema is not None:
-            result["quality_schema"] = dict(self._quality_schema)
+        if self._quality_schemas:
+            result["quality_schemas"] = list(self._quality_schemas)
 
         return result
 
@@ -315,9 +319,11 @@ def merge(self, other: "SchemaRegistry") -> None:
             if name in self._span_type_schemas:
                 conflicts.append(f"span_type_schemas/{name}")
 
-        if other._quality_schema is not None:
-            if self._quality_schema is not None:
-                conflicts.append("quality_schema")
+        if other._quality_schemas:
+            existing_names = {q["name"] for q in self._quality_schemas}
+            for entry in other._quality_schemas:
+                if entry["name"] in existing_names:
+                    conflicts.append(f"quality_schemas/{entry['name']}")
 
         if conflicts:
             msg = f"Cannot merge registries - conflicting schemas: {conflicts}"
@@ -332,8 +338,8 @@ def merge(self, other: "SchemaRegistry") -> None:
         for name, entry in other._span_type_schemas.items():
             self._span_type_schemas[name] = entry.copy()
 
-        if other._quality_schema is not None:
-            self._quality_schema = other._quality_schema.copy()
+        for entry in other._quality_schemas:
+            self._quality_schemas.append(entry.copy())
 
 
 __all__ = ["SchemaRegistry"]