feat: Implement new features and refactor core architecture

configs/pairwise_judge.yaml.example

@@ -0,0 +1,12 @@
+models:
+  - provider: openai
+    model: gpt-4-turbo
+    params:
+      temperature: 0.1
+
+rubric: |
+  You are a search quality evaluator. You will be given a query and two answers, A and B.
+  Your task is to determine which answer is better. A better answer is more relevant, accurate, and helpful.
+  Respond with a JSON object with two keys: "winner" and "reason".
+  The "winner" should be either "A" or "B".
+  The "reason" should be a brief explanation of why you chose that answer.

connectors/weaviate.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from typing import Any, Dict
+
+import aiohttp
+import weaviate
+from weaviate.auth import AuthApiKey
+
+from ..core.config import WeaviateConnectorConfig
+from ..core.dataset import Query
+
+
+class WeaviateConnector:
+    def __init__(self, cfg: WeaviateConnectorConfig) -> None:
+        self.cfg = cfg
+        auth_config = AuthApiKey(api_key=cfg.api_key) if cfg.api_key else None
+        self.client = weaviate.Client(url=cfg.url, auth_client_secret=auth_config)
+
+    async def search(self, session: aiohttp.ClientSession, query: Query) -> Dict[str, Any]:
+        del session
+        response = self.client.query.get(self.cfg.class_name, ["text", "score"]).with_near_text({"concepts": [query.text]}).do()
+        results = response.get("data", {}).get("Get", {}).get(self.cfg.class_name, [])
+        documents = [
+            {"doc_id": item.get("_additional", {}).get("id"), "text": item.get("text"), "score": item.get("_additional", {}).get("distance")}
+            for item in results
+        ]
+        return {"documents": documents}

requirements.txt

@@ -1,3 +1 @@
-altair
-pandas
-streamlit
+weaviate-client

searchqual/cli.py

@@ -1,4 +1,4 @@
-
+
 from __future__ import annotations
 
 import json
@@ -7,6 +7,9 @@ import re
 from pathlib import Path
 from typing import Dict, List, Optional
 
+import questionary
+from jinja2 import Environment, FileSystemLoader
+
 import typer
 from rich.console import Console
 from rich.table import Table
@@ -43,7 +46,21 @@ def run(
     if not dataset_path.is_absolute():
         dataset_path = (config.parent / dataset_path).resolve()
     dataset_cfg = _load_dataset_config(dataset_path)
-    runner = EvaluationRunner(run_cfg, dataset_cfg, dataset_path=dataset_path)
+
+    system = run_cfg.system
+    if system.type == 'http':
+        from .connectors.basic import HTTPSearchClient
+        connector = HTTPSearchClient(system)
+    elif system.type == 'local':
+        from .connectors.basic import LocalSearchClient
+        connector = LocalSearchClient(system)
+    elif system.type == 'weaviate':
+        from .connectors.weaviate import WeaviateConnector
+        connector = WeaviateConnector(system)
+    else:
+        raise TypeError(f'Unsupported system type: {system.type}')
+
+    runner = EvaluationRunner(run_cfg, dataset_cfg, connector, dataset_path=dataset_path)
     console.print('[bold]Running evaluation...[/bold]')
     result = runner.run()
     storage = RunStorage(Path(run_cfg.output_dir))
@@ -60,11 +77,59 @@ def run(
     _render_metrics_table(result.metrics)
 
 
+
+@app.command()
+def init():
+    """Create a new run.yaml configuration file interactively."""
+    console.print("[bold green]Welcome to the SearchQual config wizard![/bold green]")
+
+    answers = {}
+
+    def _ensure(value):
+        if value is None:
+            console.print('[yellow]Wizard cancelled. No file written.[/yellow]')
+            raise typer.Exit(code=1)
+        return value
+
+    answers['name'] = _ensure(questionary.text("Enter a name for this evaluation run:", default="my-eval-run").ask())
+    answers['dataset_path'] = _ensure(questionary.text("Enter the path to your dataset.yaml:", default="configs/dataset.yaml.example").ask())
+    answers['system_url'] = _ensure(questionary.text("Enter the URL of the search system to evaluate:", default="http://localhost:8000/search").ask())
+
+    answers['retrieval_metrics'] = questionary.checkbox(
+        "Select retrieval metrics to track:",
+        choices=["ndcg@10", "recall@100", "mrr", "map@100", "precision@10"],
+        default=["ndcg@10", "recall@100"],
+    ).ask() or []
+
+    answers['qa_metrics'] = questionary.checkbox(
+        "Select RAG/QA metrics to track:",
+        choices=["faithfulness", "citation_coverage", "nli_factuality", "helpfulness_likert"],
+        default=["faithfulness", "citation_coverage"],
+    ).ask() or []
+
+    answers['use_judge'] = questionary.confirm("Use an LLM to judge answer quality?", default=True).ask()
+    if answers['use_judge']:
+        answers['judge_provider'] = _ensure(questionary.select(
+            "Select the LLM provider for judging:",
+            choices=["openai", "bedrock", "ollama"],
+            default="openai",
+        ).ask())
+        answers['judge_model'] = _ensure(questionary.text("Enter the model name:", default="gpt-4-turbo").ask())
+
+    env = Environment(loader=FileSystemLoader(Path(__file__).parent / 'core'), trim_blocks=True, lstrip_blocks=True)
+    template = env.get_template('run.yaml.j2')
+    output_content = template.render(**answers)
+
+    output_path = Path('run.yaml')
+    output_path.write_text(output_content, encoding='utf-8')
+
+    console.print('\n[bold green]✅ Success![/bold green] Your run.yaml file has been created.')
+    console.print("You can now run an evaluation with: [cyan]sq run -c run.yaml[/cyan]")
 @app.command()
 def compare(
     baseline: Path = typer.Option(..., '--b', help='Baseline run JSON'),
     candidate: Path = typer.Option(..., '--a', help='Candidate run JSON'),
-    gate: Optional[List[str]] = typer.Option(None, '--gate', help='Gate expression, e.g. metric>=+0.01 or metric<=+5%'),
+    judge_config: Optional[Path] = typer.Option(None, '--judge', help='Path to pairwise judge config YAML'),
 ) -> None:
     """Compare two runs and display metric deltas."""
     summary = compare_runs(baseline, candidate)
@@ -83,6 +148,42 @@ def compare(
             f"{info['relative_delta']*100:+.2f}%",
         )
     console.print(table)
+
+    if judge_config:
+        from .judgers.pairwise_llm_judge import PairwiseLLMJudge
+        from .core.runner import RunResult
+        
+        console.print('\n[bold]Running pairwise evaluation...[/bold]')
+        judge_cfg = ConfigLoader.load_judge_config(judge_config)
+        judge = PairwiseLLMJudge(judge_cfg)
+        
+        baseline_run = RunResult(**json.loads(baseline.read_text()))
+        candidate_run = RunResult(**json.loads(candidate.read_text()))
+
+        baseline_results = {r['query_id']: r for r in baseline_run.results}
+        candidate_results = {r['query_id']: r for r in candidate_run.results}
+
+        wins = 0
+        losses = 0
+        ties = 0
+
+        for query_id, baseline_result in baseline_results.items():
+            if query_id in candidate_results:
+                candidate_result = candidate_results[query_id]
+                judgment = asyncio.run(judge.evaluate(baseline_result['query'], baseline_result, candidate_result))
+                if judgment:
+                    if judgment.winner == 'A':
+                        losses += 1
+                    elif judgment.winner == 'B':
+                        wins += 1
+                    else:
+                        ties += 1
+
+        console.print(f'\n[bold]Pairwise Comparison Results[/bold]')
+        console.print(f'Candidate Wins: {wins}')
+        console.print(f'Candidate Losses: {losses}')
+        console.print(f'Ties: {ties}')
+
     if gate:
         ok, failures = _evaluate_gates(summary, gate)
         if ok:
@@ -184,3 +285,5 @@ def _render_metrics_table(metrics: Dict[str, float]) -> None:
 
 if __name__ == '__main__':
     app()
+
+

searchqual/connectors/basic.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import asyncio
+import importlib
+from typing import Any, Dict
+
+import aiohttp
+
+from ..core.config import HTTPSystemConfig, LocalSystemConfig
+from ..core.dataset import Query
+
+
+class HTTPSearchClient:
+    def __init__(self, cfg: HTTPSystemConfig) -> None:
+        self.cfg = cfg
+
+    async def search(self, session: aiohttp.ClientSession, query: Query) -> Dict[str, Any]:
+        payload = {**self.cfg.params, 'query': query.text}
+        async with session.post(self.cfg.url, json=payload, headers=self.cfg.headers) as resp:
+            resp.raise_for_status()
+            return await resp.json()
+
+
+class LocalSearchClient:
+    def __init__(self, cfg: LocalSystemConfig) -> None:
+        module = importlib.import_module(cfg.module)
+        self.callable = getattr(module, cfg.object)
+        self.params = cfg.params
+
+    async def search(self, session: aiohttp.ClientSession, query: Query) -> Dict[str, Any]:
+        del session
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, self._call, query)
+
+    def _call(self, query: Query) -> Dict[str, Any]:
+        return self.callable(query.text, **self.params)

searchqual/core/config.py

@@ -46,7 +46,14 @@ class LocalSystemConfig(BaseModel):
     params: Dict[str, Any] = Field(default_factory=dict)
 
 
-SystemConfig = HTTPSystemConfig | LocalSystemConfig
+class WeaviateConnectorConfig(BaseModel):
+    type: Literal['weaviate'] = 'weaviate'
+    url: str
+    class_name: str
+    api_key: Optional[str] = None
+
+
+SystemConfig = HTTPSystemConfig | LocalSystemConfig | WeaviateConnectorConfig
 
 
 class BudgetConfig(BaseModel):
@@ -76,6 +83,11 @@ class JudgeConfig(BaseModel):
     majority_vote: bool = True
 
 
+class PairwiseJudgeConfig(BaseModel):
+    models: List[JudgeModelConfig] = Field(default_factory=list)
+    rubric: Optional[str] = None
+
+
 class RunConfig(BaseModel):
     system: SystemConfig
     dataset: str = Field(..., description="Path to dataset.yaml or dataset id")
@@ -125,6 +137,14 @@ class ConfigLoader:
         except ValidationError as exc:
             raise ConfigError(f'Invalid dataset config: {path}\n{exc}') from exc
 
+    @staticmethod
+    def load_judge_config(path: Path) -> PairwiseJudgeConfig:
+        payload = ConfigLoader.read_yaml(path)
+        try:
+            return PairwiseJudgeConfig.model_validate(payload)
+        except ValidationError as exc:
+            raise ConfigError(f'Invalid judge config: {path}\n{exc}') from exc
+
 
 __all__ = [
     'BudgetConfig',
@@ -136,6 +156,7 @@ __all__ = [
     'JudgeModelConfig',
     'LocalSystemConfig',
     'MetricSelection',
+    'PairwiseJudgeConfig',
     'RunConfig',
     'SystemConfig',
 ]

searchqual/core/connectors.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Protocol
+
+import aiohttp
+
+from .dataset import Query
+
+
+class SearchConnector(Protocol):
+    async def search(self, session: aiohttp.ClientSession, query: Query) -> Dict[str, Any]:
+        ...

searchqual/core/run.yaml.j2

@@ -0,0 +1,40 @@
+# Configuration for a SearchQual evaluation run, generated by 'sq init'
+name: {{ name }}
+dataset: {{ dataset_path }}
+output_dir: runs/
+
+system:
+  type: http
+  url: {{ system_url }}
+  # Example for other systems:
+  # type: weaviate
+  # url: "http://localhost:8080"
+  # class_name: "MyCollection"
+  # text_key: "content"
+
+metrics:
+  retrieval:
+    {%- for metric in retrieval_metrics %}
+    - {{ metric }}
+    {%- endfor %}
+  qa:
+    {%- for metric in qa_metrics %}
+    - {{ metric }}
+    {%- endfor %}
+
+{% if use_judge %}
+judge:
+  rubric: configs/rubrics/qa_v1.md
+  models:
+    - provider: {{ judge_provider }}
+      model: {{ judge_model }}
+      sample_rate: 1.0
+      params:
+        # For OpenAI, add: api_key: ${OPENAI_API_KEY}
+        # For Bedrock, add: region: us-east-1
+        temperature: 0.1
+{% endif %}
+
+budget:
+  max_tokens: 100000
+  max_dollars: 5.00

searchqual/core/runner.py

@@ -70,54 +70,18 @@ class RunResult:
         path.write_text(json.dumps(self.to_json(), indent=2), encoding='utf-8')
 
 
-class SearchClient(Protocol):
-    async def search(self, session: aiohttp.ClientSession, query: Query) -> Dict[str, Any]:
-        ...
-
-
-class HTTPSearchClient:
-    def __init__(self, cfg: HTTPSystemConfig) -> None:
-        self.cfg = cfg
-
-    async def search(self, session: aiohttp.ClientSession, query: Query) -> Dict[str, Any]:
-        payload = {**self.cfg.params, 'query': query.text}
-        async with session.post(self.cfg.url, json=payload, headers=self.cfg.headers) as resp:
-            resp.raise_for_status()
-            return await resp.json()
-
-
-class LocalSearchClient:
-    def __init__(self, cfg: LocalSystemConfig) -> None:
-        module = importlib.import_module(cfg.module)
-        self.callable = getattr(module, cfg.object)
-        self.params = cfg.params
-
-    async def search(self, session: aiohttp.ClientSession, query: Query) -> Dict[str, Any]:
-        del session
-        loop = asyncio.get_running_loop()
-        return await loop.run_in_executor(None, self._call, query)
-
-    def _call(self, query: Query) -> Dict[str, Any]:
-        return self.callable(query.text, **self.params)
+from .connectors import SearchConnector
 
 
 class EvaluationRunner:
-    def __init__(self, run_cfg: RunConfig, dataset_cfg: DatasetConfig, dataset_path: Optional[Path] = None) -> None:
+    def __init__(self, run_cfg: RunConfig, dataset_cfg: DatasetConfig, connector: SearchConnector, dataset_path: Optional[Path] = None) -> None:
         self.run_cfg = run_cfg
         self.dataset_cfg = dataset_cfg
+        self.connector = connector
         self.dataset_path = dataset_path
 
-    def _build_client(self) -> SearchClient:
-        system = self.run_cfg.system
-        if isinstance(system, HTTPSystemConfig):
-            return HTTPSearchClient(system)
-        if isinstance(system, LocalSystemConfig):
-            return LocalSearchClient(system)
-        raise TypeError(f'Unsupported system config: {system}')
-
     async def run_async(self) -> RunResult:
         dataset = self._load_dataset()
-        client = self._build_client()
         connector = aiohttp.TCPConnector(limit=self.run_cfg.system.concurrency if isinstance(self.run_cfg.system, HTTPSystemConfig) else 10)
         timeout = aiohttp.ClientTimeout(total=getattr(self.run_cfg.system, 'timeout', 30.0))
         sem = asyncio.Semaphore(getattr(self.run_cfg.system, 'concurrency', 5))
@@ -129,7 +93,7 @@ class EvaluationRunner:
             async def evaluate(query: Query) -> QueryResult:
                 async with sem:
                     start = perf_counter()
-                    response = await client.search(session, query)
+                    response = await self.connector.search(session, query)
                     latency_ms = (perf_counter() - start) * 1000
                     docs = []
                     for item in response.get('documents', response.get('results', [])):

searchqual/judgers/pairwise_llm_judge.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+import aiohttp
+
+from ..core.config import JudgeConfig, JudgeModelConfig
+from ..core.dataset import Query
+from ..core.runner import QueryResult
+
+try:
+    import boto3
+except ImportError:  # pragma: no cover - optional dependency
+    boto3 = None
+
+
+@dataclass
+class PairwiseJudgment:
+    winner: str
+    reason: str
+    raw: Dict[str, Any]
+
+
+class PairwiseLLMJudge:
+    def __init__(self, cfg: JudgeConfig) -> None:
+        self.cfg = cfg
+        self.rubric_text = self._load_rubric(cfg.rubric) if cfg.rubric else None
+
+    async def evaluate(self, query: Query, result_a: QueryResult, result_b: QueryResult) -> Optional[PairwiseJudgment]:
+        if not self.cfg.models:
+            return None
+        
+        # For now, we only use the first model for pairwise comparison
+        model_cfg = self.cfg.models[0]
+
+        async with aiohttp.ClientSession() as session:
+            return await self._call_model(session, model_cfg, query, result_a, result_b)
+
+    async def _call_model(self, session: aiohttp.ClientSession, model_cfg: JudgeModelConfig, query: Query, result_a: QueryResult, result_b: QueryResult) -> Optional[PairwiseJudgment]:
+        prompt = self._build_prompt(query, result_a, result_b)
+        if model_cfg.provider == 'openai':
+            url = model_cfg.params.get('url', 'https://api.openai.com/v1/chat/completions')
+            headers = {
+                'Authorization': f"Bearer {model_cfg.params.get('api_key', os.environ.get('OPENAI_API_KEY', ''))}",
+                'Content-Type': 'application/json',
+            }
+            payload = {
+                'model': model_cfg.model,
+                'messages': [
+                    {'role': 'system', 'content': self.rubric_text or 'You are a strict search-quality judge.'},
+                    {'role': 'user', 'content': prompt},
+                ],
+                'temperature': model_cfg.params.get('temperature', 0.1),
+            }
+            async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=60)) as resp:
+                resp.raise_for_status()
+                data = await resp.json()
+                content = data['choices'][0]['message']['content']
+                return self._parse_response(content, model_cfg)
+        raise NotImplementedError(f'Provider {model_cfg.provider} not implemented')
+
+    def _parse_response(self, text: str, model_cfg: JudgeModelConfig) -> Optional[PairwiseJudgment]:
+        try:
+            data = json.loads(text)
+            winner = data.get('winner')
+            reason = data.get('reason')
+            if winner in ['A', 'B']:
+                return PairwiseJudgment(winner=winner, reason=reason, raw=data)
+        except json.JSONDecodeError:
+            pass
+        return None
+
+    def _build_prompt(self, query: Query, result_a: QueryResult, result_b: QueryResult) -> str:
+        rubric = self.rubric_text or 'Compare the two answers and determine which one is better.'
+        prompt_lines = [
+            rubric,
+            '',
+            f'Query: {query.text}',
+            '',
+            'Answer A:',
+            result_a.answer,
+            '',
+            'Answer B:',
+            result_b.answer,
+            '',
+            'Which answer is better? Respond with JSON with keys "winner" (either "A" or "B") and "reason".',
+        ]
+        return '\n'.join(line for line in prompt_lines if line)
+
+    def _load_rubric(self, path: str) -> str:
+        return Path(path).read_text(encoding='utf-8')
+
+
+__all__ = ['PairwiseLLMJudge']

src/streamlit_app.py

@@ -1,40 +0,0 @@
-import altair as alt
-import numpy as np
-import pandas as pd
-import streamlit as st
-
-"""
-# Welcome to Streamlit!
-
-Edit `/streamlit_app.py` to customize this app to your heart's desire :heart:.
-If you have any questions, checkout our [documentation](https://docs.streamlit.io) and [community
-forums](https://discuss.streamlit.io).
-
-In the meantime, below is an example of what you can do with just a few lines of code:
-"""
-
-num_points = st.slider("Number of points in spiral", 1, 10000, 1100)
-num_turns = st.slider("Number of turns in spiral", 1, 300, 31)
-
-indices = np.linspace(0, 1, num_points)
-theta = 2 * np.pi * num_turns * indices
-radius = indices
-
-x = radius * np.cos(theta)
-y = radius * np.sin(theta)
-
-df = pd.DataFrame({
-    "x": x,
-    "y": y,
-    "idx": indices,
-    "rand": np.random.randn(num_points),
-})
-
-st.altair_chart(alt.Chart(df, height=700, width=700)
-    .mark_point(filled=True)
-    .encode(
-        x=alt.X("x", axis=None),
-        y=alt.Y("y", axis=None),
-        color=alt.Color("idx", legend=None, scale=alt.Scale()),
-        size=alt.Size("rand", legend=None, scale=alt.Scale(range=[1, 150])),
-    ))

streamlit_app.py

@@ -42,8 +42,26 @@ with col1:
         selected = st.selectbox('Select run', run_files, format_func=lambda p: p.stem)
         if st.button('Load run'):
             data = json.loads(selected.read_text())
-            st.json(data.get('metrics', {}))
-            st.write('Stats', data.get('stats', {}))
+            metrics = data.get('metrics', {})
+            if metrics:
+                st.subheader("Metrics")
+                cols = st.columns(len(metrics))
+                for i, (k, v) in enumerate(metrics.items()):
+                    cols[i].metric(k, v)
+
+            stats = data.get('stats', {})
+            if stats:
+                st.subheader("Stats")
+                st.table(stats)
+
+            results = data.get('results', [])
+            if results:
+                st.subheader("Results")
+                for res in results:
+                    st.write(f"**Query:** {res['query']}")
+                    if res.get('answer'):
+                        st.write(f"**Answer:** {res['answer']}")
+                    st.dataframe(res.get('documents', []))
     else:
         st.info('No runs found yet. Execute a run to populate metrics.')