DAG Evaluator

Dockerfile.jupyter

@@ -1,13 +1,14 @@
 FROM python:3.12-slim
 
 ENV PATH="/root/.local/bin:$PATH"
+ENV PYTHONPATH="/app/flightpaths/flights"
 # Used for the notebook server
 WORKDIR /app
 
 # pipx needed for uv installation script
 # ssh client needed for installing private modelbench dependencies
 # git needed dvc
-RUN apt-get update && apt-get install -y pipx openssh-client git && \
+RUN apt-get update && apt-get install -y pipx openssh-client git graphviz && \
     pipx install uv
 COPY pyproject.toml uv.lock README.md ./
 

pyproject.toml

@@ -23,6 +23,7 @@ dependencies = [
     "scikit-learn>=1.5.0,<2.0.0",
     "pandas>=2.2.2,<4",
     "modelbench @ git+https://github.com/mlcommons/modelbench.git",
+    "graphviz>=0.20,<1",
 ]
 
 [project.scripts]

src/modelplane/evaluator/annotator.py

@@ -0,0 +1,32 @@
+from modelgauge.annotator import Annotator
+from modelgauge.prompt import ChatPrompt, TextPrompt
+from modelgauge.prompt_formatting import format_chat
+from modelgauge.sut import SUTResponse
+
+from modelplane.evaluator.context import EvalContext
+from modelplane.evaluator.dag import EvaluatorDAG
+from modelplane.evaluator.outputs import Output
+
+
+class DAGAnnotator(Annotator):
+    """Annotator that executes a DAG."""
+
+    def __init__(self, uid: str, dag: EvaluatorDAG) -> None:
+        super().__init__(uid)
+        self.dag = dag
+
+    def translate_prompt(
+        self,
+        prompt: TextPrompt | ChatPrompt,
+        response: SUTResponse,
+    ) -> EvalContext:
+        prompt_str = (
+            prompt.text if isinstance(prompt, TextPrompt) else format_chat(prompt)
+        )
+        return EvalContext(
+            prompt=prompt_str,
+            response=response.text,
+        )
+
+    def annotate(self, annotation_request: EvalContext) -> Output:
+        return self.dag.run(annotation_request)

src/modelplane/evaluator/context.py

@@ -0,0 +1,20 @@
+from typing import Any, Optional
+
+
+class EvalContext:
+    """Context state passed around during DAG execution."""
+
+    def __init__(
+        self, prompt: str, response: str, metadata: Optional[dict[str, Any]] = None
+    ) -> None:
+        self.prompt = prompt
+        self.response = response
+        self.metadata = metadata or {}
+        self._parent_outputs = {}
+
+    def set_parent_outputs(self, outputs: dict[str, Any]) -> None:
+        self._parent_outputs = outputs
+
+    def parent_outputs(self) -> list[Any]:
+        """Return the NodeOutput for a specific node, or None if it was skipped."""
+        return list(self._parent_outputs.values())