feat: pluggable LLM provider interface for judge

fleet/__init__.py

@@ -74,6 +74,9 @@
 # Import judge data classes
 from .judge import Rubric, Criterion, File, Image, JudgeResult
 
+# Import LLM provider interface
+from .llm_provider import LLMProvider, FleetProvider, ExternalProvider, resolve_provider
+
 # Create a module-level env attribute for convenient access
 from . import env
 from . import global_client as _global_client
@@ -102,6 +105,11 @@
     "File",
     "Image",
     "JudgeResult",
+    # LLM Providers
+    "LLMProvider",
+    "FleetProvider",
+    "ExternalProvider",
+    "resolve_provider",
     # Exceptions
     "FleetError",
     "FleetAPIError",

fleet/_async/client.py

@@ -55,6 +55,7 @@
 if TYPE_CHECKING:
     from .verifiers import AsyncVerifierFunction
     from .judge import AsyncJudge
+    from ..llm_provider import LLMProvider
 
 
 def _json_default(x: Any) -> Any:
@@ -340,12 +341,19 @@ def message_count(self) -> int:
 
 
 class AsyncEnv(EnvironmentBase):
-    def __init__(self, client: Optional[AsyncWrapper], **kwargs):
+    def __init__(
+        self,
+        client: Optional[AsyncWrapper],
+        *,
+        llm_provider: Optional["LLMProvider"] = None,
+        **kwargs,
+    ):
         super().__init__(**kwargs)
         self._client = client
         self._apps: Dict[str, AsyncInstanceClient] = {}
         self._instance: Optional[AsyncInstanceClient] = None
         self._judge: Optional["AsyncJudge"] = None
+        self._llm_provider = llm_provider
 
     @property
     def instance(self) -> AsyncInstanceClient:
@@ -423,13 +431,18 @@ def mcp(self) -> AsyncMCPResource:
 
     @property
     def judge(self) -> "AsyncJudge":
-        """LLM-as-judge grading via orchestrator API."""
+        """LLM-as-judge grading.
+
+        Routes through Fleet orchestrator by default. Set ``llm_provider``
+        on the environment to route to an external provider instead.
+        """
         if self._judge is None:
             from .judge import AsyncJudge
 
             self._judge = AsyncJudge(
-                client=self._load_client,
+                client=self._client,
                 instance_id=self.instance_id,
+                llm_provider=self._llm_provider,
             )
         return self._judge
 

fleet/_async/judge.py

@@ -1,9 +1,15 @@
 """Fleet SDK Judge - Async version.
 
 Provides env.judge.grade() for async verifier scripts.
+
+Provider resolution order:
+
+1. Explicit ``llm_provider`` kwarg (highest priority)
+2. ``FLEET_LLM_API_KEY`` env var → auto-builds ``ExternalProvider``
+3. Fleet orchestrator (default fallback)
 """
 
-from typing import Dict, List, Optional, Union, TYPE_CHECKING
+from typing import Any, Dict, List, Optional, Union, TYPE_CHECKING
 
 # Import shared classes and helpers from the sync module
 from ..judge import (
@@ -19,10 +25,12 @@
     _guess_media_type,
     _parse_grade_response,
     _print_judge_call_start,
+    _UNSET,
 )
 
 if TYPE_CHECKING:
     from .base import AsyncWrapper
+    from ..llm_provider import LLMProvider
 
 # Re-export data classes so `from fleet._async.judge import ...` works
 __all__ = [
@@ -36,15 +44,33 @@
 
 
 class AsyncJudge:
-    """LLM-as-judge grading — calls orchestrator API, not environment API.
+    """LLM-as-judge grading (async).
+
+    Accessed as ``env.judge`` on AsyncEnv instances.
 
-    Accessed as env.judge on AsyncEnv instances.
+    Provider resolution order:
+
+    1. Explicit ``llm_provider`` kwarg (highest priority)
+    2. ``FLEET_LLM_API_KEY`` env var → auto-builds ``ExternalProvider``
+    3. Fleet orchestrator (default fallback)
     """
 
-    def __init__(self, client: "AsyncWrapper", instance_id: str):
+    def __init__(
+        self,
+        client: Optional["AsyncWrapper"],
+        instance_id: str,
+        *,
+        llm_provider: Any = _UNSET,
+    ):
         self._client = client
         self._instance_id = instance_id
 
+        if llm_provider is _UNSET:
+            from ..llm_provider import resolve_provider
+            self._llm_provider = resolve_provider()
+        else:
+            self._llm_provider = llm_provider
+
     async def grade(
         self,
         rubric: Union[str, Rubric],
@@ -63,7 +89,11 @@ async def grade(
         collect: Optional[Dict[str, List[str]]] = None,
         task_id: Optional[str] = None,
     ) -> JudgeResult:
-        """Grade a submission using LLM-as-judge via the orchestrator API.
+        """Grade a submission using LLM-as-judge.
+
+        Routes through the Fleet orchestrator by default. If an
+        ``llm_provider`` was set at construction time, calls the external
+        provider directly instead.
 
         Returns a JudgeResult (float subclass with .details, .criteria, .feedback)
         that can be returned directly from a verifier function.
@@ -84,59 +114,100 @@ async def grade(
             collect: File patterns for orchestrator to collect (agentic mode).
             task_id: Optional task ID for tracking.
         """
-        # Resolve Image.from_env images asynchronously before building request
+        # Fold reference_claims into context
+        effective_context = context
+        if reference_claims is not None:
+            if effective_context:
+                effective_context = f"{effective_context}\n\n## Reference Claims\n{reference_claims}"
+            else:
+                effective_context = f"## Reference Claims\n{reference_claims}"
+
+        # Resolve path-based images/files through the provider first
         resolved_images = images
-        if images and not agentic:
-            resolved_images = {}
-            for label, img in images.items():
+        resolved_files = files
+        if self._llm_provider is not None:
+            resolved_images = self._llm_provider.resolve_images(images)
+            resolved_files = self._llm_provider.resolve_files(files)
+
+        # Resolve Image.from_env images asynchronously before building request
+        if resolved_images and not agentic:
+            env_resolved = {}
+            for label, img in resolved_images.items():
                 if img.source == "env" and img._env is not None:
                     b64 = await _collect_image_from_env_async(img._env, img.filename)
                     if b64 is not None:
-                        resolved_images[label] = Image.from_base64(
+                        env_resolved[label] = Image.from_base64(
                             b64,
                             img.filename or "image.png",
                             _guess_media_type(img.filename or "image.png"),
                         )
                     else:
                         # Async collection failed — use collect source directly
                         # (don't keep the env image or serialize() will retry sync)
-                        resolved_images[label] = Image(
+                        env_resolved[label] = Image(
                             source="collect",
                             filename=img.filename,
                         )
                 else:
-                    resolved_images[label] = img
+                    env_resolved[label] = img
+            resolved_images = env_resolved
 
         # Resolve File.from_env files asynchronously before building request
-        resolved_files = files
-        if files and not agentic:
-            resolved_files = {}
-            for label, f in files.items():
+        if resolved_files and not agentic:
+            env_resolved_files = {}
+            for label, f in resolved_files.items():
                 if f.source == "env" and f._env is not None:
                     b64 = await _collect_file_from_env_async(f._env, f.filename)
                     if b64 is not None:
-                        resolved_files[label] = File.from_base64(
+                        env_resolved_files[label] = File.from_base64(
                             b64,
                             f.filename or "file",
                             _guess_file_media_type(f.filename or "file"),
                         )
                     else:
                         # Async collection failed — use collect source directly
-                        resolved_files[label] = File(
+                        env_resolved_files[label] = File(
                             source="collect",
                             filename=f.filename,
                         )
                 else:
-                    resolved_files[label] = f
+                    env_resolved_files[label] = f
+            resolved_files = env_resolved_files
 
+        _print_judge_call_start(rubric, resolved_images, agentic, model, files=resolved_files)
+
+        if self._llm_provider is not None:
+            # Route through pluggable LLM provider
+            from ..llm_provider import GradeRequest
+
+            request = GradeRequest(
+                rubric=rubric,
+                submission=submission,
+                ground_truth=ground_truth,
+                problem=problem,
+                context=effective_context,
+                conversation=conversation,
+                images=resolved_images,
+                files=resolved_files,
+                model=model,
+                provider=provider,
+                agentic=agentic,
+                collect=collect,
+                task_id=task_id,
+                instance_id=self._instance_id,
+            )
+            grade_response = await self._llm_provider.agrade(request)
+            return _parse_grade_response(grade_response.to_dict())
+
+        # Default: route through Fleet orchestrator
         body = _build_grade_request(
             self._instance_id,
             rubric,
             submission,
             ground_truth=ground_truth,
             problem=problem,
-            context=context,
-            reference_claims=reference_claims,
+            context=effective_context,
+            reference_claims=None,  # already folded into context
             conversation=conversation,
             images=resolved_images,
             files=resolved_files,
@@ -147,6 +218,5 @@ async def grade(
             task_id=task_id,
         )
 
-        _print_judge_call_start(rubric, resolved_images, agentic, model, files=resolved_files)
         response = await self._client.request("POST", "/v1/judge/grade", json=body)
         return _parse_grade_response(response.json())

fleet/client.py

@@ -60,6 +60,7 @@
 if TYPE_CHECKING:
     from .verifiers import SyncVerifierFunction
     from .judge import SyncJudge
+    from .llm_provider import LLMProvider
 
 
 def _json_default(x: Any) -> Any:
@@ -344,12 +345,19 @@ def message_count(self) -> int:
 
 
 class SyncEnv(EnvironmentBase):
-    def __init__(self, client: Optional[SyncWrapper], **kwargs):
+    def __init__(
+        self,
+        client: Optional[SyncWrapper],
+        *,
+        llm_provider: Optional["LLMProvider"] = None,
+        **kwargs,
+    ):
         super().__init__(**kwargs)
         self._client = client
         self._apps: Dict[str, InstanceClient] = {}
         self._instance: Optional[InstanceClient] = None
         self._judge: Optional["SyncJudge"] = None
+        self._llm_provider = llm_provider
         self._manager_url_override: Optional[str] = None  # For URL mode
 
     @property
@@ -435,13 +443,18 @@ def mcp(self) -> SyncMCPResource:
 
     @property
     def judge(self) -> "SyncJudge":
-        """LLM-as-judge grading via orchestrator API."""
+        """LLM-as-judge grading.
+
+        Routes through Fleet orchestrator by default. Set ``llm_provider``
+        on the environment to route to an external provider instead.
+        """
         if self._judge is None:
             from .judge import SyncJudge
 
             self._judge = SyncJudge(
-                client=self._load_client,
+                client=self._client,
                 instance_id=self.instance_id,
+                llm_provider=self._llm_provider,
             )
         return self._judge