Add simple evaluate_item endpoint

Signed-off-by: Anuradha Karuppiah <[email protected]>

Author: AnuradhaKaruppiah

src/nat/front_ends/fastapi/fastapi_front_end_config.py

@@ -27,6 +27,8 @@
 from nat.data_models.component_ref import ObjectStoreRef
 from nat.data_models.front_end import FrontEndBaseConfig
 from nat.data_models.step_adaptor import StepAdaptorConfig
+from nat.eval.evaluator.evaluator_model import EvalInputItem
+from nat.eval.evaluator.evaluator_model import EvalOutputItem
 
 logger = logging.getLogger(__name__)
 
@@ -133,6 +135,19 @@ class AsyncGenerationStatusResponse(BaseAsyncStatusResponse):
         description="Output of the generate request, this is only available if the job completed successfully.")
 
 
+class EvaluateItemRequest(BaseModel):
+    """Request model for single-item evaluation endpoint."""
+    item: EvalInputItem = Field(description="Single evaluation input item to evaluate")
+    evaluator_name: str = Field(description="Name of the evaluator to use (must match config)")
+
+
+class EvaluateItemResponse(BaseModel):
+    """Response model for single-item evaluation endpoint."""
+    success: bool = Field(description="Whether the evaluation completed successfully")
+    result: EvalOutputItem | None = Field(default=None, description="Evaluation result if successful")
+    error: str | None = Field(default=None, description="Error message if evaluation failed")
+
+
 class FastApiFrontEndConfig(FrontEndBaseConfig, name="fastapi"):
     """
     A FastAPI based front end that allows a NAT workflow to be served as a microservice.
@@ -232,6 +247,13 @@ class CrossOriginResourceSharing(BaseModel):
         description="Evaluates the performance and accuracy of the workflow on a dataset",
     )
 
+    evaluate_item: typing.Annotated[EndpointBase,
+                                    Field(description="Endpoint for evaluating a single item.")] = EndpointBase(
+                                        method="POST",
+                                        path="/evaluate_item",
+                                        description="Evaluate a single item with a specified evaluator",
+                                    )
+
     oauth2_callback_path: str | None = Field(
         default="/auth/redirect",
         description="OAuth2.0 authentication callback endpoint. If None, no OAuth2 callback endpoint is created.")

src/nat/front_ends/fastapi/fastapi_front_end_plugin_worker.py

@@ -39,6 +39,8 @@
 from pydantic import Field
 from starlette.websockets import WebSocket
 
+from nat.builder.eval_builder import WorkflowEvalBuilder
+from nat.builder.evaluator import EvaluatorInfo
 from nat.builder.function import Function
 from nat.builder.workflow_builder import WorkflowBuilder
 from nat.data_models.api_server import ChatRequest
@@ -51,11 +53,14 @@
 from nat.eval.config import EvaluationRunOutput
 from nat.eval.evaluate import EvaluationRun
 from nat.eval.evaluate import EvaluationRunConfig
+from nat.eval.evaluator.evaluator_model import EvalInput
 from nat.front_ends.fastapi.auth_flow_handlers.http_flow_handler import HTTPAuthenticationFlowHandler
 from nat.front_ends.fastapi.auth_flow_handlers.websocket_flow_handler import FlowState
 from nat.front_ends.fastapi.auth_flow_handlers.websocket_flow_handler import WebSocketAuthenticationFlowHandler
 from nat.front_ends.fastapi.fastapi_front_end_config import AsyncGenerateResponse
 from nat.front_ends.fastapi.fastapi_front_end_config import AsyncGenerationStatusResponse
+from nat.front_ends.fastapi.fastapi_front_end_config import EvaluateItemRequest
+from nat.front_ends.fastapi.fastapi_front_end_config import EvaluateItemResponse
 from nat.front_ends.fastapi.fastapi_front_end_config import EvaluateRequest
 from nat.front_ends.fastapi.fastapi_front_end_config import EvaluateResponse
 from nat.front_ends.fastapi.fastapi_front_end_config import EvaluateStatusResponse
@@ -227,6 +232,50 @@ def __init__(self, config: Config):
         self._outstanding_flows: dict[str, FlowState] = {}
         self._outstanding_flows_lock = asyncio.Lock()
 
+        # Evaluator storage for single-item evaluation
+        self._evaluators: dict[str, EvaluatorInfo] = {}
+        self._eval_builder: WorkflowEvalBuilder | None = None
+
+    async def initialize_evaluators(self, config: Config):
+        """Initialize and store evaluators from config for single-item evaluation."""
+        if not config.eval or not config.eval.evaluators:
+            logger.info("No evaluators configured, skipping evaluator initialization")
+            return
+
+        try:
+            # Build evaluators using WorkflowEvalBuilder
+            self._eval_builder = WorkflowEvalBuilder(general_config=config.general,
+                                                     eval_general_config=config.eval.general,
+                                                     registry=None)
+
+            # Enter the async context and keep it alive
+            await self._eval_builder.__aenter__()
+
+            # Populate evaluators
+            for name, evaluator_config in config.eval.evaluators.items():
+                await self._eval_builder.add_evaluator(name, evaluator_config)
+                self._evaluators[name] = self._eval_builder.get_evaluator(name)
+                logger.info(f"Initialized evaluator: {name}")
+
+            logger.info(f"Successfully initialized {len(self._evaluators)} evaluators")
+
+        except Exception as e:
+            logger.error(f"Failed to initialize evaluators: {e}")
+            # Don't fail startup, just log the error
+            self._evaluators = {}
+
+    async def cleanup_evaluators(self):
+        """Clean up evaluator resources on shutdown."""
+        if self._eval_builder:
+            try:
+                await self._eval_builder.__aexit__(None, None, None)
+                logger.info("Evaluator builder context cleaned up")
+            except Exception as e:
+                logger.error(f"Error cleaning up evaluator builder: {e}")
+            finally:
+                self._eval_builder = None
+                self._evaluators.clear()
+
     def get_step_adaptor(self) -> StepAdaptor:
 
         return StepAdaptor(self.front_end_config.step_adaptor)
@@ -236,12 +285,16 @@ async def configure(self, app: FastAPI, builder: WorkflowBuilder):
         # Do things like setting the base URL and global configuration options
         app.root_path = self.front_end_config.root_path
 
+        # Initialize evaluators for single-item evaluation
+        await self.initialize_evaluators(self._config)
+
         await self.add_routes(app, builder)
 
     async def add_routes(self, app: FastAPI, builder: WorkflowBuilder):
 
         await self.add_default_route(app, SessionManager(await builder.build()))
         await self.add_evaluate_route(app, SessionManager(await builder.build()))
+        await self.add_evaluate_item_route(app, SessionManager(await builder.build()))
         await self.add_static_files_route(app, builder)
         await self.add_authorization_route(app)
         await self.add_mcp_client_tool_list_route(app, builder)
@@ -439,6 +492,69 @@ async def get_jobs(http_request: Request, status: str | None = None) -> list[Eva
             else:
                 logger.warning("Dask is not available, evaluation endpoints will not be added.")
 
+    async def add_evaluate_item_route(self, app: FastAPI, session_manager: SessionManager):
+        """Add the single-item evaluation endpoint to the FastAPI app."""
+
+        async def evaluate_single_item(request: EvaluateItemRequest, http_request: Request) -> EvaluateItemResponse:
+            """Handle single-item evaluation requests."""
+
+            async with session_manager.session(http_connection=http_request):
+
+                # Check if evaluator exists
+                if request.evaluator_name not in self._evaluators:
+                    raise HTTPException(status_code=404,
+                                        detail=f"Evaluator '{request.evaluator_name}' not found. "
+                                        f"Available evaluators: {list(self._evaluators.keys())}")
+
+                try:
+                    # Get the evaluator
+                    evaluator = self._evaluators[request.evaluator_name]
+
+                    # Run evaluation on single item
+                    result = await evaluator.evaluate_fn(EvalInput(eval_input_items=[request.item]))
+
+                    # Extract the single output item
+                    if result.eval_output_items:
+                        output_item = result.eval_output_items[0]
+                        return EvaluateItemResponse(success=True, result=output_item, error=None)
+                    else:
+                        return EvaluateItemResponse(success=False, result=None, error="Evaluator returned no results")
+
+                except Exception as e:
+                    logger.exception(f"Error evaluating item with {request.evaluator_name}")
+                    return EvaluateItemResponse(success=False, result=None, error=f"Evaluation failed: {str(e)}")
+
+        # Register the route
+        if self.front_end_config.evaluate_item.path:
+            app.add_api_route(path=self.front_end_config.evaluate_item.path,
+                              endpoint=evaluate_single_item,
+                              methods=[self.front_end_config.evaluate_item.method],
+                              response_model=EvaluateItemResponse,
+                              description=self.front_end_config.evaluate_item.description,
+                              responses={
+                                  404: {
+                                      "description": "Evaluator not found",
+                                      "content": {
+                                          "application/json": {
+                                              "example": {
+                                                  "detail": "Evaluator 'unknown' not found"
+                                              }
+                                          }
+                                      }
+                                  },
+                                  500: {
+                                      "description": "Internal Server Error",
+                                      "content": {
+                                          "application/json": {
+                                              "example": {
+                                                  "detail": "Internal server error occurred"
+                                              }
+                                          }
+                                      }
+                                  }
+                              })
+            logger.info(f"Added evaluate_item route at {self.front_end_config.evaluate_item.path}")
+
     async def add_static_files_route(self, app: FastAPI, builder: WorkflowBuilder):
 
         if not self.front_end_config.object_store: