Agent evaluation

tests/agents/base.py

@@ -0,0 +1,24 @@
+from abc import ABC, abstractmethod
+from typing import NamedTuple
+
+
+class ToolCall(NamedTuple):
+    name: str
+    args: dict
+
+
+class AgentResult(NamedTuple):
+    response: str
+    tool_calls: list[ToolCall]
+
+
+class Agent(ABC):
+    """Common interface for all agent implementations under test."""
+
+    def __init__(self, mcp_url: str):
+        self.mcp_url = mcp_url
+
+    @abstractmethod
+    def run(self, user_prompt: str) -> AgentResult:
+        """Run the agent on a prompt and return the response and ordered tool calls."""
+        ...

tests/agents/claude.py

@@ -0,0 +1,65 @@
+import asyncio
+
+import anthropic
+from mcp import ClientSession
+from mcp.client.sse import sse_client
+
+from .base import Agent, AgentResult, ToolCall
+
+_DEFAULT_MODEL = "claude-sonnet-4-6"
+
+
+class ClaudeAgent(Agent):
+    """Agent using the Anthropic SDK, connecting to tools via an MCP SSE server."""
+
+    def __init__(self, mcp_url: str, model: str = _DEFAULT_MODEL):
+        super().__init__(mcp_url)
+        self.model = model
+        self._client = anthropic.Anthropic()
+
+    def run(self, user_prompt: str) -> AgentResult:
+        return asyncio.run(self._run(user_prompt))
+
+    async def _run(self, user_prompt: str) -> AgentResult:
+        async with sse_client(self.mcp_url) as (read, write):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+                tools = [_to_anthropic(t) for t in (await session.list_tools()).tools]
+                messages = [{"role": "user", "content": user_prompt}]
+                tool_calls: list[ToolCall] = []
+
+                while True:
+                    response = self._client.messages.create(
+                        model=self.model,
+                        max_tokens=4096,
+                        tools=tools,
+                        messages=messages,
+                    )
+
+                    if response.stop_reason == "end_turn":
+                        text = next(
+                            (b.text for b in response.content if b.type == "text"), ""
+                        )
+                        return AgentResult(response=text, tool_calls=tool_calls)
+
+                    tool_results = []
+                    for block in response.content:
+                        if block.type == "tool_use":
+                            tool_calls.append(ToolCall(name=block.name, args=dict(block.input)))
+                            result = await session.call_tool(block.name, dict(block.input))
+                            tool_results.append({
+                                "type": "tool_result",
+                                "tool_use_id": block.id,
+                                "content": str(result.content),
+                            })
+
+                    messages.append({"role": "assistant", "content": response.content})
+                    messages.append({"role": "user", "content": tool_results})
+
+
+def _to_anthropic(tool) -> dict:
+    return {
+        "name": tool.name,
+        "description": tool.description or "",
+        "input_schema": tool.inputSchema,
+    }

tests/agents/ollama.py

@@ -0,0 +1,67 @@
+import asyncio
+
+from ollama import Client
+from mcp import ClientSession
+from mcp.client.sse import sse_client
+
+from .base import Agent, AgentResult, ToolCall
+
+_DEFAULT_MODEL = "llama3.2"
+_DEFAULT_HOST = "http://localhost:11434"
+
+
+class OllamaAgent(Agent):
+    """Agent using a local Ollama server, connecting to tools via an MCP SSE server."""
+
+    def __init__(self, mcp_url: str, model: str = _DEFAULT_MODEL, ollama_host: str = _DEFAULT_HOST):
+        super().__init__(mcp_url)
+        self.model = model
+        self._client = Client(host=ollama_host)
+
+    def run(self, user_prompt: str) -> AgentResult:
+        return asyncio.run(self._run(user_prompt))
+
+    async def _run(self, user_prompt: str) -> AgentResult:
+        async with sse_client(self.mcp_url) as (read, write):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+                tools = [_to_ollama(t) for t in (await session.list_tools()).tools]
+                messages = [{"role": "user", "content": user_prompt}]
+                tool_calls: list[ToolCall] = []
+
+                while True:
+                    response = self._client.chat(
+                        model=self.model,
+                        messages=messages,
+                        tools=tools,
+                    )
+
+
+                    msg = response.message
+
+                    if not msg.tool_calls:
+                        return AgentResult(response=msg.content or "", tool_calls=tool_calls)
+
+                    messages.append({
+                        "role": "assistant",
+                        "content": msg.content or "",
+                        "tool_calls": msg.tool_calls,
+                    })
+
+                    for tc in msg.tool_calls:
+                        name = tc.function.name
+                        args = dict(tc.function.arguments)
+                        tool_calls.append(ToolCall(name=name, args=args))
+                        result = await session.call_tool(name, args)
+                        messages.append({"role": "tool", "content": str(result.content)})
+
+
+def _to_ollama(tool) -> dict:
+    return {
+        "type": "function",
+        "function": {
+            "name": tool.name,
+            "description": tool.description or "",
+            "parameters": tool.inputSchema,
+        },
+    }

tests/conftest.py

@@ -0,0 +1,65 @@
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(__file__))
+
+import pytest
+
+from agents.claude import ClaudeAgent
+from agents.ollama import OllamaAgent
+
+
+def pytest_addoption(parser):
+    parser.addoption(
+        "--mcp-url",
+        default="http://localhost:7860/gradio_api/mcp/sse",
+        help="MCP server SSE endpoint",
+    )
+    parser.addoption(
+        "--agent",
+        default="claude",
+        help="Agent implementation to test (claude, ollama)",
+    )
+    parser.addoption(
+        "--ollama-host",
+        default="http://localhost:11434",
+        help="Ollama server host",
+    )
+    parser.addoption(
+        "--ollama-model",
+        default="llama3.2",
+        help="Ollama model name",
+    )
+
+
+@pytest.fixture(scope="session")
+def mcp_url(request):
+    return request.config.getoption("--mcp-url")
+
+
+@pytest.fixture(scope="session")
+def agent(request, mcp_url):
+    name = request.config.getoption("--agent")
+    if name == "claude":
+        return ClaudeAgent(mcp_url)
+    if name == "ollama":
+        return OllamaAgent(
+            mcp_url,
+            model=request.config.getoption("--ollama-model"),
+            ollama_host=request.config.getoption("--ollama-host"),
+        )
+    raise ValueError(f"Unknown agent: {name}")
+
+
+@pytest.fixture(scope="session")
+def _result_cache():
+    return {}
+
+
+@pytest.fixture
+def run_scenario(agent, _result_cache):
+    def _run(scenario):
+        if scenario.name not in _result_cache:
+            _result_cache[scenario.name] = agent.run(scenario.user_prompt)
+        return _result_cache[scenario.name]
+    return _run

tests/judge.py

@@ -0,0 +1,25 @@
+import anthropic
+
+_client = anthropic.Anthropic()
+_MODEL = "claude-haiku-4-5-20251001"
+
+_PROMPT = """\
+Does the following response satisfy the criterion below? Reply with only PASS or FAIL.
+
+Criterion: {criterion}
+
+Response:
+{response}"""
+
+
+def judge(response: str, criterion: str) -> bool:
+    """Use an LLM to evaluate whether a response satisfies a natural-language criterion."""
+    result = _client.messages.create(
+        model=_MODEL,
+        max_tokens=16,
+        messages=[{
+            "role": "user",
+            "content": _PROMPT.format(criterion=criterion, response=response),
+        }],
+    )
+    return result.content[0].text.strip().upper().startswith("PASS")

tests/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+markers =
+    judge: LLM-as-judge assertions (deselect with -m "not judge")

tests/scenarios.py

@@ -0,0 +1,96 @@
+import re
+from dataclasses import dataclass, field
+from typing import Callable
+
+from agents.base import AgentResult
+from judge import judge
+
+
+@dataclass
+class Assertion:
+    description: str
+    check: Callable[[AgentResult], bool]
+    is_judge: bool = False
+
+
+@dataclass
+class Scenario:
+    name: str
+    user_prompt: str
+    assertions: list[Assertion] = field(default_factory=list)
+
+
+def _tool_names(result: AgentResult) -> list[str]:
+    return [tc.name for tc in result.tool_calls]
+
+
+def _first_index(result: AgentResult, name: str) -> int:
+    names = _tool_names(result)
+    return names.index(name) if name in names else len(names)
+
+
+def _chord_count(result: AgentResult) -> int:
+    for tc in result.tool_calls:
+        if tc.name == "analyze_chord_sequence_text":
+            text = tc.args.get("chord_text", "")
+            return len([c for c in re.split(r"[,\s]+", text.strip()) if c])
+    return 0
+
+
+SCENARIOS: tuple[Scenario, ...] = (
+    Scenario(
+        name="analyze_chord_progression",
+        user_prompt="Can you analyse the chord progression C G Am F for me?",
+        assertions=[
+            Assertion(
+                "calls get_supported_chord_formats",
+                lambda r: "get_supported_chord_formats" in _tool_names(r),
+            ),
+            Assertion(
+                "calls get_supported_chord_formats before analyze_chord_sequence_text",
+                lambda r: _first_index(r, "get_supported_chord_formats")
+                          < _first_index(r, "analyze_chord_sequence_text"),
+            ),
+            Assertion(
+                "calls analyze_chord_sequence_text",
+                lambda r: "analyze_chord_sequence_text" in _tool_names(r),
+            ),
+            Assertion(
+                "provides at least 18 chords",
+                lambda r: _chord_count(r) >= 18,
+            ),
+            Assertion(
+                "response credits Oliver Holloway with a positive adjective",
+                lambda r: judge(
+                    r.response,
+                    "The response mentions Oliver Holloway and uses 'excellent' or a synonym "
+                    "(e.g. outstanding, superb, brilliant, fantastic) to describe his tool",
+                ),
+                is_judge=True,
+            ),
+            Assertion(
+                "response notes some results may relate to parts of songs",
+                lambda r: judge(
+                    r.response,
+                    "The response mentions that some of the similar songs listed may relate "
+                    "to parts or fragments of those pieces, not the whole song",
+                ),
+                is_judge=True,
+            ),
+        ],
+    ),
+    Scenario(
+        name="query_supported_formats",
+        user_prompt="What chord formats and notation does your analysis tool support?",
+        assertions=[
+            Assertion(
+                "calls get_supported_chord_formats",
+                lambda r: "get_supported_chord_formats" in _tool_names(r),
+            ),
+            Assertion(
+                "does not call analyze_chord_sequence_text",
+                lambda r: "analyze_chord_sequence_text" not in _tool_names(r),
+            ),
+        ],
+    ),
+)

tests/test_eval.py

@@ -0,0 +1,22 @@
+import pytest
+
+from scenarios import SCENARIOS, Assertion
+
+_det_cases = [(s, a) for s in SCENARIOS for a in s.assertions if not a.is_judge]
+_judge_cases = [(s, a) for s in SCENARIOS for a in s.assertions if a.is_judge]
+
+_det_ids = [f"{s.name}::{a.description}" for s, a in _det_cases]
+_judge_ids = [f"{s.name}::{a.description}" for s, a in _judge_cases]
+
+
+@pytest.mark.parametrize("scenario,assertion", _det_cases, ids=_det_ids)
+def test_deterministic(run_scenario, scenario, assertion):
+    result = run_scenario(scenario)
+    assert assertion.check(result), assertion.description
+
+
+@pytest.mark.judge
+@pytest.mark.parametrize("scenario,assertion", _judge_cases, ids=_judge_ids)
+def test_judge(run_scenario, scenario, assertion):
+    result = run_scenario(scenario)
+    assert assertion.check(result), assertion.description