feat: add judge evaluation support to agent graphs

Adds per-node judge evaluation to agent graph execution. Each AIAgentConfig
now carries a pre-built Evaluator (mirroring AICompletionConfig) that the
provider-specific AgentGraphRunner invokes after each node's model response.
Results are tracked via the same AIConfigTracker used for that node's LLM
metrics, ensuring evaluation data is correlated correctly.

Key changes:
- New Evaluator class coordinating multiple judges; evaluate() returns an
  asyncio Task so evaluation fires immediately and is awaited in flush()
- AIAgentConfig and AICompletionConfig carry an eager evaluator (kw_only field)
- LangGraph runner stores per-node eval tasks in _pending_eval_tasks and
  flushes them via the callback handler's async flush() method
- OpenAI runner fires judge evaluation at handoff and final-segment points
- client._build_evaluator() handles empty/None judge config via Evaluator.noop()

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: jsonbailey

packages/ai-providers/server-ai-langchain/src/ldai_langchain/langchain_runner_factory.py

@@ -39,7 +39,11 @@ def create_agent(self, config: Any, tools: Optional[ToolRegistry] = None) -> Lan
         )
         return LangChainAgentRunner(agent)
 
-    def create_agent_graph(self, graph_def: Any, tools: ToolRegistry) -> Any:
+    def create_agent_graph(
+        self,
+        graph_def: Any,
+        tools: ToolRegistry,
+    ) -> Any:
         """
         CAUTION:
         This feature is experimental and should NOT be considered ready for production use.

packages/ai-providers/server-ai-langchain/src/ldai_langchain/langgraph_agent_graph_runner.py

@@ -1,7 +1,8 @@
 """LangGraph agent graph runner for LaunchDarkly AI SDK."""
 
+import asyncio
 import time
-from typing import Annotated, Any, Dict, List, Optional, Set, Tuple
+from typing import Annotated, Any, Dict, List, Set, Tuple
 
 from ldai import log
 from ldai.agent_graph import AgentGraphDefinition, AgentGraphNode
@@ -67,7 +68,11 @@ class LangGraphAgentGraphRunner(AgentGraphRunner):
     Requires ``langgraph`` to be installed.
     """
 
-    def __init__(self, graph: AgentGraphDefinition, tools: ToolRegistry):
+    def __init__(
+        self,
+        graph: AgentGraphDefinition,
+        tools: ToolRegistry,
+    ):
         """
         Initialize the runner.
 
@@ -79,6 +84,7 @@ def __init__(self, graph: AgentGraphDefinition, tools: ToolRegistry):
         self._compiled: Any = None
         self._fn_name_to_config_key: Dict[str, str] = {}
         self._node_keys: Set[str] = set()
+        self._pending_eval_tasks: Dict[str, asyncio.Task] = {}
 
     def _ensure_compiled(self) -> None:
         """Build and cache the compiled graph if not already done."""
@@ -172,6 +178,18 @@ async def invoke(state: WorkflowState) -> dict:
                     if node_instructions:
                         msgs = [SystemMessage(content=node_instructions)] + msgs
                     response = await bound_model.ainvoke(msgs)
+
+                    node_obj = self._graph.get_node(nk)
+                    if node_obj is not None:
+                        input_text = '\r\n'.join(
+                            m.content if isinstance(m.content, str) else str(m.content)
+                            for m in msgs
+                        ) if msgs else ''
+                        output_text = (
+                            response.content if hasattr(response, 'content') else str(response)
+                        )
+                        self._pending_eval_tasks[nk] = node_obj.get_config().evaluator.evaluate(input_text, output_text)
+
                     return {'messages': [response]}
 
                 invoke.__name__ = nk
@@ -280,6 +298,7 @@ async def run(self, input: Any) -> AgentGraphResult:
         :param input: The string prompt to send to the agent graph
         :return: AgentGraphResult with the final output and metrics
         """
+        self._pending_eval_tasks = {}
         tracker = self._graph.create_tracker() if self._graph.create_tracker is not None else None
         start_ns = time.perf_counter_ns()
 
@@ -299,7 +318,7 @@ async def run(self, input: Any) -> AgentGraphResult:
             output = extract_last_message_content(messages)
 
             # Flush per-node metrics to LD trackers
-            handler.flush(self._graph)
+            await handler.flush(self._graph, self._pending_eval_tasks)
 
             # Graph-level metrics
             if tracker:

packages/ai-providers/server-ai-langchain/src/ldai_langchain/langgraph_callback_handler.py

@@ -188,13 +188,15 @@ def on_tool_end(
     # Flush
     # ------------------------------------------------------------------
 
-    def flush(self, graph: AgentGraphDefinition) -> None:
+    async def flush(self, graph: AgentGraphDefinition, eval_tasks=None) -> None:
         """
         Emit all collected per-node metrics to the LaunchDarkly trackers.
 
         Call this once after the graph run completes.
 
         :param graph: The AgentGraphDefinition whose nodes hold the LD config trackers.
+        :param eval_tasks: Optional dict mapping node key to an awaitable that returns
+            judge evaluation results.
         """
         node_trackers: Dict[str, Any] = {}
         for node_key in self._path:
@@ -220,3 +222,15 @@ def flush(self, graph: AgentGraphDefinition) -> None:
 
             for tool_key in self._node_tool_calls.get(node_key, []):
                 config_tracker.track_tool_call(tool_key)
+
+            if not eval_tasks:
+                continue
+
+            eval_task = eval_tasks.get(node_key)
+            if not eval_task:
+                continue
+
+            results = await eval_task
+            for r in results:
+                if r.success:
+                    config_tracker.track_judge_result(r)

packages/ai-providers/server-ai-langchain/tests/test_langchain_provider.py

@@ -6,6 +6,7 @@
 from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
 
 from ldai import LDMessage
+from ldai.evaluator import Evaluator
 
 from ldai_langchain import (
     LangChainModelRunner,
@@ -530,6 +531,7 @@ def sync_tool(x: str = '') -> str:
         cfg = AIAgentConfig(
             key='n',
             enabled=True,
+            evaluator=Evaluator.noop(),
             create_tracker=MagicMock(),
             model=ModelConfig(
                 name='gpt-4',
@@ -553,6 +555,7 @@ async def async_tool(x: str = '') -> str:
         cfg = AIAgentConfig(
             key='n',
             enabled=True,
+            evaluator=Evaluator.noop(),
             create_tracker=MagicMock(),
             model=ModelConfig(
                 name='gpt-4',

packages/ai-providers/server-ai-langchain/tests/test_langgraph_agent_graph_runner.py

@@ -4,6 +4,7 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 from ldai.agent_graph import AgentGraphDefinition
+from ldai.evaluator import Evaluator
 from ldai.models import AIAgentGraphConfig, AIAgentConfig, ModelConfig, ProviderConfig
 from ldai.providers import AgentGraphResult, ToolRegistry
 from ldai_langchain.langgraph_agent_graph_runner import LangGraphAgentGraphRunner
@@ -20,6 +21,7 @@ def _make_graph(enabled: bool = True) -> AgentGraphDefinition:
         model=ModelConfig(name='gpt-4'),
         provider=ProviderConfig(name='openai'),
         instructions='You are a helpful assistant.',
+        evaluator=Evaluator.noop(),
     )
     graph_config = AIAgentGraphConfig(
         key='test-graph',

packages/ai-providers/server-ai-langchain/tests/test_langgraph_callback_handler.py

@@ -17,6 +17,7 @@
 from ldai.agent_graph import AgentGraphDefinition
 from ldai.models import AIAgentConfig, AIAgentGraphConfig, ModelConfig, ProviderConfig
 from ldai.tracker import AIGraphTracker, LDAIConfigTracker, TokenUsage
+from ldai.evaluator import Evaluator
 from ldai_langchain.langgraph_callback_handler import LDMetricsCallbackHandler
 
 
@@ -48,6 +49,7 @@ def _make_graph(mock_ld_client: MagicMock, node_key: str = 'root-agent', graph_k
     node_config = AIAgentConfig(
         key=node_key,
         enabled=True,
+        evaluator=Evaluator.noop(),
         model=ModelConfig(name='gpt-4', parameters={}),
         provider=ProviderConfig(name='openai'),
         instructions='Be helpful.',
@@ -317,7 +319,8 @@ def test_on_tool_end_none_name_ignored():
 # flush() tests
 # ---------------------------------------------------------------------------
 
-def test_flush_emits_token_events_to_ld_tracker():
+@pytest.mark.asyncio
+async def test_flush_emits_token_events_to_ld_tracker():
     """flush() calls track_tokens on the node's config tracker."""
     mock_ld_client = MagicMock()
     graph = _make_graph(mock_ld_client, node_key='root-agent', graph_key='g1')
@@ -327,7 +330,7 @@ def test_flush_emits_token_events_to_ld_tracker():
     node_run_id = uuid4()
     handler.on_chain_start({}, {}, run_id=node_run_id, name='root-agent')
     handler.on_llm_end(_llm_result(15, 10, 5), run_id=uuid4(), parent_run_id=node_run_id)
-    handler.flush(graph)
+    await handler.flush(graph)
 
     ev = _events(mock_ld_client)
     assert ev['$ld:ai:tokens:total'][0][1] == 15
@@ -336,7 +339,8 @@ def test_flush_emits_token_events_to_ld_tracker():
     assert ev['$ld:ai:generation:success'][0][1] == 1
 
 
-def test_flush_emits_duration():
+@pytest.mark.asyncio
+async def test_flush_emits_duration():
     """flush() calls track_duration when duration was recorded."""
     mock_ld_client = MagicMock()
     graph = _make_graph(mock_ld_client)
@@ -346,13 +350,14 @@ def test_flush_emits_duration():
     run_id = uuid4()
     handler.on_chain_start({}, {}, run_id=run_id, name='root-agent')
     handler.on_chain_end({}, run_id=run_id)
-    handler.flush(graph)
+    await handler.flush(graph)
 
     ev = _events(mock_ld_client)
     assert '$ld:ai:duration:total' in ev
 
 
-def test_flush_emits_tool_calls():
+@pytest.mark.asyncio
+async def test_flush_emits_tool_calls():
     """flush() calls track_tool_call for each recorded tool invocation."""
     mock_ld_client = MagicMock()
     graph = _make_graph(mock_ld_client)
@@ -366,15 +371,16 @@ def test_flush_emits_tool_calls():
     tools_run_id = uuid4()
     handler.on_chain_start({}, {}, run_id=tools_run_id, name='root-agent__tools')
     handler.on_tool_end('r', run_id=uuid4(), parent_run_id=tools_run_id, name='fn_search')
-    handler.flush(graph)
+    await handler.flush(graph)
 
     ev = _events(mock_ld_client)
     tool_events = ev.get('$ld:ai:tool_call', [])
     assert len(tool_events) == 1
     assert tool_events[0][0]['toolKey'] == 'search'
 
 
-def test_flush_includes_graph_key_in_node_events():
+@pytest.mark.asyncio
+async def test_flush_includes_graph_key_in_node_events():
     """flush() passes graph_key to the node tracker so graphKey appears in events."""
     mock_ld_client = MagicMock()
     graph = _make_graph(mock_ld_client, graph_key='my-graph')
@@ -384,14 +390,15 @@ def test_flush_includes_graph_key_in_node_events():
     node_run_id = uuid4()
     handler.on_chain_start({}, {}, run_id=node_run_id, name='root-agent')
     handler.on_llm_end(_llm_result(5, 3, 2), run_id=uuid4(), parent_run_id=node_run_id)
-    handler.flush(graph)
+    await handler.flush(graph)
 
     ev = _events(mock_ld_client)
     token_data = ev['$ld:ai:tokens:total'][0][0]
     assert token_data.get('graphKey') == 'my-graph'
 
 
-def test_flush_with_no_graph_key_on_node_tracker():
+@pytest.mark.asyncio
+async def test_flush_with_no_graph_key_on_node_tracker():
     """When node tracker has no graph_key, events omit graphKey."""
     mock_ld_client = MagicMock()
     context = MagicMock()
@@ -408,6 +415,7 @@ def test_flush_with_no_graph_key_on_node_tracker():
     node_config = AIAgentConfig(
         key='root-agent',
         enabled=True,
+        evaluator=Evaluator.noop(),
         model=ModelConfig(name='gpt-4', parameters={}),
         provider=ProviderConfig(name='openai'),
         instructions='Be helpful.',
@@ -432,29 +440,31 @@ def test_flush_with_no_graph_key_on_node_tracker():
     node_run_id = uuid4()
     handler.on_chain_start({}, {}, run_id=node_run_id, name='root-agent')
     handler.on_llm_end(_llm_result(5, 3, 2), run_id=uuid4(), parent_run_id=node_run_id)
-    handler.flush(graph)
+    await handler.flush(graph)
 
     ev = _events(mock_ld_client)
     token_data = ev['$ld:ai:tokens:total'][0][0]
     assert 'graphKey' not in token_data
 
 
-def test_flush_skips_nodes_not_in_path():
+@pytest.mark.asyncio
+async def test_flush_skips_nodes_not_in_path():
     """flush() only emits events for nodes that were actually executed."""
     mock_ld_client = MagicMock()
     graph = _make_graph(mock_ld_client)
     tracker = graph.create_tracker()
 
     # Handler with 'root-agent' in node_keys but never started
     handler = LDMetricsCallbackHandler({'root-agent'}, {})
-    handler.flush(graph)
+    await handler.flush(graph)
 
     ev = _events(mock_ld_client)
     assert '$ld:ai:tokens:total' not in ev
     assert '$ld:ai:generation:success' not in ev
 
 
-def test_flush_skips_node_without_tracker():
+@pytest.mark.asyncio
+async def test_flush_skips_node_without_tracker():
     """flush() silently skips nodes whose config has no tracker."""
     mock_ld_client = MagicMock()
     context = MagicMock()
@@ -463,6 +473,7 @@ def test_flush_skips_node_without_tracker():
         key='no-track',
         enabled=True,
         create_tracker=lambda: None,
+        evaluator=Evaluator.noop(),
         model=ModelConfig(name='gpt-4', parameters={}),
         provider=ProviderConfig(name='openai'),
         instructions='',
@@ -483,7 +494,7 @@ def test_flush_skips_node_without_tracker():
     node_run_id = uuid4()
     handler.on_chain_start({}, {}, run_id=node_run_id, name='no-track')
     handler.on_llm_end(_llm_result(5, 3, 2), run_id=uuid4(), parent_run_id=node_run_id)
-    handler.flush(graph)  # should not raise
+    await handler.flush(graph)  # should not raise
 
     mock_ld_client.track.assert_not_called()