From ace71b0fd054a8a3a0e958bb484635b81571358e Mon Sep 17 00:00:00 2001 From: Johannes du Plessis Date: Wed, 6 May 2026 10:15:58 -0700 Subject: [PATCH] feat: add reviewer graph + eval target wiring (#1241) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: add reviewer graph + eval target wiring - New `reviewer` graph (`agent/reviewer.py`) registered in langgraph.json alongside the main `agent` graph. Reuses the same sandbox lifecycle, GH proxy auth, and middleware primitives from `agent.server`, but with a narrower tool set, a reviewer-specific system prompt, no commit/push, and the `task` (subagent) tool stripped via `_ToolExclusionMiddleware` so review stays in one context. - New `github_comment` tool: agents call it once per issue with `(file, line, body, severity)` and the eval scores those calls against golden comments. - `ensure_no_empty_msg` middleware (the no_op nudge) is intentionally *not* on the reviewer's stack — that middleware exists to enforce the main agent's "always finalize via Slack/Linear/PR" contract, which the reviewer doesn't have. The main agent's behavior is unchanged. - `evals/reviewer/target.py`: send PR info as a user message, extract every `github_comment` tool call (multiple expected per review) into the run output. - `evals/reviewer/judge.py`: per-example evaluator now returns a list of metrics under `{"results": [...]}` so LangSmith averages each numeric key (f1/precision/recall/tp/fp/fn) across the experiment in the UI. Dropped the broken `aggregate_pr` summary evaluator that reached for an attribute that doesn't exist on `RunTree`. - `evals/reviewer/run_eval.py`: `--limit` now slices the dataset via `client.list_examples(limit=N)` since `aevaluate` doesn't accept `max_examples`. - Makefile: `dev` and `run` targets now use `uv run` so they work without an activated venv. * resolve comments --------- Co-authored-by: open-swe[bot] --- Makefile | 4 +- agent/middleware/__init__.py | 2 + agent/middleware/exclude_tools.py | 65 ++++++++++++++ agent/reviewer.py | 132 ++++++++++++++++++++++++++++ agent/server.py | 141 +++++++++++++++++------------- agent/tools/__init__.py | 2 + agent/tools/github_comment.py | 53 +++++++++++ evals/reviewer/judge.py | 139 ++++++++++++++++------------- evals/reviewer/run_eval.py | 62 ++++++++++--- evals/reviewer/target.py | 107 ++++++++++++++--------- langgraph.json | 3 +- 11 files changed, 532 insertions(+), 178 deletions(-) create mode 100644 agent/middleware/exclude_tools.py create mode 100644 agent/reviewer.py create mode 100644 agent/tools/github_comment.py diff --git a/Makefile b/Makefile index 666594e1..d1268071 100644 --- a/Makefile +++ b/Makefile @@ -8,10 +8,10 @@ all: help ###################### dev: - langgraph dev + uv run langgraph dev run: - uvicorn agent.webapp:app --reload --port 8000 + uv run uvicorn agent.webapp:app --reload --port 8000 install: uv pip install -e . diff --git a/agent/middleware/__init__.py b/agent/middleware/__init__.py index 52bcd458..da26db6b 100644 --- a/agent/middleware/__init__.py +++ b/agent/middleware/__init__.py @@ -1,10 +1,12 @@ from .check_message_queue import check_message_queue_before_model from .ensure_no_empty_msg import ensure_no_empty_msg +from .exclude_tools import ExcludeToolsMiddleware from .notify_step_limit import notify_step_limit_reached from .sanitize_tool_inputs import SanitizeToolInputsMiddleware from .tool_error_handler import ToolErrorMiddleware __all__ = [ + "ExcludeToolsMiddleware", "SanitizeToolInputsMiddleware", "ToolErrorMiddleware", "check_message_queue_before_model", diff --git a/agent/middleware/exclude_tools.py b/agent/middleware/exclude_tools.py new file mode 100644 index 00000000..d45f8992 --- /dev/null +++ b/agent/middleware/exclude_tools.py @@ -0,0 +1,65 @@ +"""Hide named tools from the model without rebuilding the agent. + +`create_deep_agent` always wires the `task` tool when the auto-added +general-purpose subagent is present. The reviewer agent has no use for +subagent dispatch, so this middleware drops the named tools from the +request before the model sees them. Mirrors the behavior of deepagents' +own private `_ToolExclusionMiddleware` but lives here so we don't depend +on a private import path. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from typing import Any + +from langchain.agents.middleware.types import ( + AgentMiddleware, + AgentState, + ModelRequest, + ModelResponse, +) +from langchain_core.tools import BaseTool + + +def _tool_name(tool: BaseTool | dict[str, Any] | Any) -> str | None: + if isinstance(tool, dict): + name = tool.get("name") + return name if isinstance(name, str) else None + name = getattr(tool, "name", None) + return name if isinstance(name, str) else None + + +class ExcludeToolsMiddleware(AgentMiddleware): + """Strip named tools from each model request. + + Place this AFTER tool-injecting middleware (FilesystemMiddleware, + SubAgentMiddleware) so it can remove middleware-injected tools too. + """ + + state_schema = AgentState + + def __init__(self, *, excluded: frozenset[str]) -> None: + self._excluded = excluded + + def _filter(self, request: ModelRequest) -> ModelRequest: + if not self._excluded: + return request + filtered = [t for t in request.tools if _tool_name(t) not in self._excluded] + if len(filtered) == len(request.tools): + return request + return request.override(tools=filtered) + + def wrap_model_call( + self, + request: ModelRequest, + handler: Callable[[ModelRequest], ModelResponse], + ) -> ModelResponse: + return handler(self._filter(request)) + + async def awrap_model_call( + self, + request: ModelRequest, + handler: Callable[[ModelRequest], Awaitable[ModelResponse]], + ) -> ModelResponse: + return await handler(self._filter(request)) diff --git a/agent/reviewer.py b/agent/reviewer.py new file mode 100644 index 00000000..cc41a7be --- /dev/null +++ b/agent/reviewer.py @@ -0,0 +1,132 @@ +"""Reviewer graph factory. + +Mirrors `agent.server.get_agent`'s sandbox lifecycle but returns a deep agent +configured for code review only: narrowed tool set, reviewer-specific system +prompt, no commit/push/PR-opening. + +Inline review comments are recorded by the agent calling the `github_comment` +tool — one call per distinct issue. The eval harness extracts those calls +from the run's message stream. +""" +# ruff: noqa: E402 + +import logging +import os +import warnings + +logger = logging.getLogger(__name__) + +from langgraph.graph.state import RunnableConfig +from langgraph.pregel import Pregel + +warnings.filterwarnings("ignore", module="langchain_core._api.deprecation") +warnings.filterwarnings("ignore", message=".*Pydantic V1.*", category=UserWarning) + +from deepagents import create_deep_agent +from langchain.agents.middleware import ModelCallLimitMiddleware + +from .middleware import ( + ExcludeToolsMiddleware, + SanitizeToolInputsMiddleware, + ToolErrorMiddleware, +) +from .server import ( + DEFAULT_LLM_MAX_TOKENS, + DEFAULT_LLM_MODEL_ID, + DEFAULT_LLM_REASONING, + DEFAULT_RECURSION_LIMIT, + MODEL_CALL_RECURSION_LIMIT, + ensure_sandbox_for_thread, + graph_loaded_for_execution, +) +from .tools import github_comment +from .utils.auth import resolve_github_token +from .utils.model import ModelKwargs, make_model +from .utils.sandbox_paths import aresolve_sandbox_work_dir + +REVIEWER_PROMPT_TEMPLATE = """You are an expert code reviewer. + +Your job is to review a single GitHub pull request and surface real issues — +bugs, security problems, correctness errors, race conditions, performance +regressions, and clear quality issues. Do not nitpick style. + +### Working environment + +You are operating in a remote Linux sandbox at `{working_dir}`. + +- The `gh` CLI is installed and authenticated by a sandbox proxy. Always + invoke it as `GH_TOKEN=dummy gh `. +- The `execute` tool runs shell commands. The default timeout is generous + (~30 minutes); pass `timeout=` only if you need to override it. + +### How to review + +1. The user message tells you which PR to review (URL, repo, PR number, + base SHA, head SHA). +2. Clone the repo into `{working_dir}` and check out the **base SHA** so + the working tree matches `main` at the time the PR was opened. +3. Fetch the PR head and inspect the diff: + `GH_TOKEN=dummy gh pr diff --repo /` + or use `git diff ...`. +4. Read the files the PR changes — and any related files needed to + understand the change in context. Use `read_file`, `grep`, `glob`. +5. For each real issue you find, call the `github_comment` tool **once** + with: + - `file`: repo-relative path + - `line`: 1-based line number in the new (post-PR) file + - `body`: a specific description of the issue + - `severity`: one of "Low", "Medium", "High", "Critical" + +### Hard rules + +- **You are read-only.** Do NOT commit. Do NOT push. Do NOT open or update + PRs. Do NOT post comments via `gh pr comment`. The only way you record + findings is by calling the `github_comment` tool. +- One `github_comment` call per distinct issue. Multiple calls per review + are expected and correct. +- Do not summarize the PR in chat. Do not write a final review essay. + Only `github_comment` calls are scored — anything else is ignored. +- If you find no real issues, make zero `github_comment` calls and stop. +""" + + +def _reviewer_system_prompt(working_dir: str) -> str: + return REVIEWER_PROMPT_TEMPLATE.format(working_dir=working_dir) + + +async def get_reviewer_agent(config: RunnableConfig) -> Pregel: + """Get or create a reviewer agent with a sandbox for the given thread.""" + thread_id = config["configurable"].get("thread_id", None) + + config["recursion_limit"] = DEFAULT_RECURSION_LIMIT + + if thread_id is None or not graph_loaded_for_execution(config): + logger.info("No thread_id or not for execution, returning reviewer agent without sandbox") + return create_deep_agent(system_prompt="", tools=[]).with_config(config) + + if config["configurable"].get("source"): + _token, new_encrypted = await resolve_github_token(config, thread_id) + config["metadata"]["github_token_encrypted"] = new_encrypted + del _token + + sandbox_backend = await ensure_sandbox_for_thread(thread_id) + + work_dir = await aresolve_sandbox_work_dir(sandbox_backend) + + model_id = os.environ.get("LLM_MODEL_ID", DEFAULT_LLM_MODEL_ID) + model_kwargs: ModelKwargs = {"max_tokens": DEFAULT_LLM_MAX_TOKENS} + if model_id == DEFAULT_LLM_MODEL_ID: + model_kwargs["reasoning"] = DEFAULT_LLM_REASONING + + return create_deep_agent( + model=make_model(model_id, **model_kwargs), + system_prompt=_reviewer_system_prompt(work_dir), + tools=[github_comment], + backend=sandbox_backend, + middleware=[ + SanitizeToolInputsMiddleware(), + ModelCallLimitMiddleware(run_limit=MODEL_CALL_RECURSION_LIMIT, exit_behavior="end"), + ToolErrorMiddleware(), + ExcludeToolsMiddleware(excluded=frozenset({"task"})), + ], + ).with_config(config) diff --git a/agent/server.py b/agent/server.py index 652147c8..6404fe7d 100644 --- a/agent/server.py +++ b/agent/server.py @@ -175,6 +175,82 @@ def graph_loaded_for_execution(config: RunnableConfig) -> bool: ) +async def ensure_sandbox_for_thread(thread_id: str) -> SandboxBackendProtocol: + """Get-or-create a healthy sandbox bound to ``thread_id``. + + Implements the four-state lifecycle described in AGENTS.md: + + 1. Cached in memory → ping; recreate on ``SandboxClientError``. + 2. Metadata says ``__creating__`` and no cache → poll until ready. + 3. No sandbox at all → create one and persist the id. + 4. Metadata has an id but no cache → reconnect; recreate on failure. + + For LangSmith sandboxes, also refreshes the GitHub App proxy auth. + Persists the resulting ``sandbox_id`` to thread metadata, and on the + first creation/reconnect for this thread initializes git identity. + """ + sandbox_backend = SANDBOX_BACKENDS.get(thread_id) + sandbox_id = await get_sandbox_id_from_metadata(thread_id) + + if sandbox_id == SANDBOX_CREATING and not sandbox_backend: + logger.info("Sandbox creation in progress for thread %s, waiting...", thread_id) + sandbox_id = await _wait_for_sandbox_id(thread_id) + + if sandbox_backend: + logger.info("Using cached sandbox backend for thread %s", thread_id) + await _refresh_github_proxy(sandbox_backend) + sandbox_backend = await check_or_recreate_sandbox(sandbox_backend, thread_id) + elif sandbox_id is None: + logger.info("Creating new sandbox for thread %s", thread_id) + await client.threads.update(thread_id=thread_id, metadata={"sandbox_id": SANDBOX_CREATING}) + try: + sandbox_backend = await _create_sandbox_with_proxy() + logger.info("Sandbox created: %s", sandbox_backend.id) + except Exception: + logger.exception("Failed to create sandbox") + try: + await client.threads.update(thread_id=thread_id, metadata={"sandbox_id": None}) + except Exception: + logger.exception("Failed to reset sandbox_id metadata") + raise + else: + logger.info("Connecting to existing sandbox %s", sandbox_id) + try: + sandbox_backend = await asyncio.to_thread(create_sandbox, sandbox_id) + except Exception: + logger.warning("Failed to connect to existing sandbox %s, creating new one", sandbox_id) + await client.threads.update( + thread_id=thread_id, metadata={"sandbox_id": SANDBOX_CREATING} + ) + try: + sandbox_backend = await _create_sandbox_with_proxy() + except Exception: + logger.exception("Failed to create replacement sandbox") + await client.threads.update(thread_id=thread_id, metadata={"sandbox_id": None}) + raise + await _refresh_github_proxy(sandbox_backend) + sandbox_backend = await check_or_recreate_sandbox(sandbox_backend, thread_id) + + SANDBOX_BACKENDS[thread_id] = sandbox_backend + + if sandbox_id != sandbox_backend.id: + await client.threads.update( + thread_id=thread_id, metadata={"sandbox_id": sandbox_backend.id} + ) + + # Re-apply git identity every run: cached/reconnected sandboxes may have + # lost their `--global` config (or had it overwritten), and Vercel preview + # deploys reject commits whose author email can't be resolved to a GitHub + # account. + await asyncio.to_thread( + sandbox_backend.execute, + "git config --global user.name 'open-swe[bot]' && " + "git config --global user.email 'open-swe@users.noreply.github.com'", + ) + + return sandbox_backend + + DEFAULT_LLM_MODEL_ID = "openai:gpt-5.5" DEFAULT_LLM_REASONING: OpenAIReasoning = {"effort": "medium"} DEFAULT_LLM_MAX_TOKENS = 64_000 @@ -197,70 +273,9 @@ async def get_agent(config: RunnableConfig) -> Pregel: github_token, new_encrypted = await resolve_github_token(config, thread_id) config["metadata"]["github_token_encrypted"] = new_encrypted + del github_token - sandbox_backend = SANDBOX_BACKENDS.get(thread_id) - sandbox_id = await get_sandbox_id_from_metadata(thread_id) - - if sandbox_id == SANDBOX_CREATING and not sandbox_backend: - logger.info("Sandbox creation in progress, waiting...") - sandbox_id = await _wait_for_sandbox_id(thread_id) - - if sandbox_backend: - logger.info("Using cached sandbox backend for thread %s", thread_id) - await _refresh_github_proxy(sandbox_backend) - sandbox_backend = await check_or_recreate_sandbox(sandbox_backend, thread_id) - - elif sandbox_id is None: - logger.info("Creating new sandbox for thread %s", thread_id) - await client.threads.update(thread_id=thread_id, metadata={"sandbox_id": SANDBOX_CREATING}) - - try: - sandbox_backend = await _create_sandbox_with_proxy() - logger.info("Sandbox created: %s", sandbox_backend.id) - except Exception: - logger.exception("Failed to create sandbox") - try: - await client.threads.update(thread_id=thread_id, metadata={"sandbox_id": None}) - logger.info("Reset sandbox_id to None for thread %s", thread_id) - except Exception: - logger.exception("Failed to reset sandbox_id metadata") - raise - else: - logger.info("Connecting to existing sandbox %s", sandbox_id) - try: - sandbox_backend = await asyncio.to_thread(create_sandbox, sandbox_id) - logger.info("Connected to existing sandbox %s", sandbox_id) - except Exception: - logger.warning("Failed to connect to existing sandbox %s, creating new one", sandbox_id) - # Reset sandbox_id and create a new sandbox with proxy auth configured - await client.threads.update( - thread_id=thread_id, - metadata={"sandbox_id": SANDBOX_CREATING}, - ) - - try: - sandbox_backend = await _create_sandbox_with_proxy() - logger.info("New sandbox created: %s", sandbox_backend.id) - except Exception: - logger.exception("Failed to create replacement sandbox") - await client.threads.update(thread_id=thread_id, metadata={"sandbox_id": None}) - raise - - await _refresh_github_proxy(sandbox_backend) - sandbox_backend = await check_or_recreate_sandbox(sandbox_backend, thread_id) - - SANDBOX_BACKENDS[thread_id] = sandbox_backend - - if sandbox_id != sandbox_backend.id: - await client.threads.update( - thread_id=thread_id, - metadata={"sandbox_id": sandbox_backend.id}, - ) - - await asyncio.to_thread( - sandbox_backend.execute, - "git config --global user.name 'open-swe[bot]' && git config --global user.email 'open-swe@users.noreply.github.com'", - ) + sandbox_backend = await ensure_sandbox_for_thread(thread_id) linear_issue = config["configurable"].get("linear_issue", {}) linear_project_id = linear_issue.get("linear_project_id", "") diff --git a/agent/tools/__init__.py b/agent/tools/__init__.py index c1857857..f868c50b 100644 --- a/agent/tools/__init__.py +++ b/agent/tools/__init__.py @@ -1,4 +1,5 @@ from .fetch_url import fetch_url +from .github_comment import github_comment from .http_request import http_request from .linear_comment import linear_comment from .linear_create_issue import linear_create_issue @@ -13,6 +14,7 @@ from .web_search import web_search __all__ = [ "fetch_url", + "github_comment", "http_request", "linear_comment", "linear_create_issue", diff --git a/agent/tools/github_comment.py b/agent/tools/github_comment.py new file mode 100644 index 00000000..1dfbbe1a --- /dev/null +++ b/agent/tools/github_comment.py @@ -0,0 +1,53 @@ +from typing import Any, Literal + +Severity = Literal["Low", "Medium", "High", "Critical"] + +_VALID_SEVERITIES: frozenset[str] = frozenset({"Low", "Medium", "High", "Critical"}) + + +def _normalize_severity(value: str) -> Severity: + """Title-case `value` and validate it against the allowed set. + + The model occasionally emits "low"/"HIGH" instead of the title-cased + canonical form. Normalize before recording so we don't burn an LLM + turn on a Pydantic ValidationError retry. + """ + titled = value.strip().title() + if titled not in _VALID_SEVERITIES: + valid = ", ".join(sorted(_VALID_SEVERITIES)) + raise ValueError(f"severity must be one of {valid}; got {value!r}") + return titled # type: ignore[return-value] + + +def github_comment( + file: str, + line: int, + body: str, + severity: str, +) -> dict[str, Any]: + """Record a single inline review comment on the PR under review. + + Call this tool once per issue you find. Multiple calls are expected — one + per distinct concern. The eval harness records every github_comment call + you make and scores them against the PR's golden comments. + + **Do not** use this tool to summarize the PR or make general remarks. Each + call must point at a specific file and line and describe one concrete + issue (bug, security concern, perf problem, correctness issue, etc.). + + Args: + file: Repo-relative path to the file the comment applies to. + line: 1-based line number in the file. + body: The review comment text. Be specific about the issue. + severity: One of "Low", "Medium", "High", "Critical" (case-insensitive). + + Returns: + {"recorded": True, "file", "line", "severity", "body"}. + """ + return { + "recorded": True, + "file": file, + "line": line, + "severity": _normalize_severity(severity), + "body": body, + } diff --git a/evals/reviewer/judge.py b/evals/reviewer/judge.py index 0eab82e5..a6505a4e 100644 --- a/evals/reviewer/judge.py +++ b/evals/reviewer/judge.py @@ -2,7 +2,8 @@ Pairwise matches each agent-emitted candidate against each golden comment using claude-opus-4-5 (the model martian used to score Devin Review). Returns -precision/recall/f1 per example, plus aggregate metrics across the experiment. +precision/recall/f1 per example, plus aggregate micro/macro metrics across +the experiment via a summary evaluator. The judge prompt is kept verbatim from withmartian/code-review-benchmark `step3_judge_comments.py` so scores are @@ -12,10 +13,11 @@ directly comparable to martian's published numbers. from __future__ import annotations import json +import threading from typing import Any +from uuid import UUID from langchain_anthropic import ChatAnthropic -from langsmith.evaluation import EvaluationResult from langsmith.schemas import Example, Run JUDGE_MODEL = "claude-opus-4-5" @@ -87,28 +89,46 @@ def _judge_pair(golden: dict, candidate: dict) -> dict[str, Any]: return {"match": False, "confidence": 0.0, "reasoning": f"unparseable: {raw[:200]}"} -def judge_match(run: Run, example: Example) -> EvaluationResult: - """Per-example evaluator: compute precision/recall/f1 against golden comments.""" +_PER_EXAMPLE_COUNTS: dict[UUID, dict[str, int | float]] = {} +_COUNTS_LOCK = threading.Lock() + + +def _record_counts(example_id: UUID, counts: dict[str, int | float]) -> None: + with _COUNTS_LOCK: + _PER_EXAMPLE_COUNTS[example_id] = counts + + +def _drain_counts() -> list[dict[str, int | float]]: + with _COUNTS_LOCK: + snapshot = list(_PER_EXAMPLE_COUNTS.values()) + _PER_EXAMPLE_COUNTS.clear() + return snapshot + + +def judge_match(run: Run, example: Example) -> dict[str, Any]: + """Per-example evaluator: compute precision/recall/f1/tp/fp/fn against goldens. + + Stashes the raw counts on a process-local cache keyed by ``example.id`` so + ``aggregate_pr`` can compute micro-averages without re-judging. + """ candidates: list[dict] = list((run.outputs or {}).get("comments") or []) goldens: list[dict] = list((example.outputs or {}).get("golden_comments") or []) if not goldens: - return EvaluationResult(key="f1", score=None, comment="no goldens") + return {"results": [{"key": "f1", "score": None, "comment": "no goldens"}]} matched_goldens: set[int] = set() matched_candidates: set[int] = set() - pair_results: list[dict] = [] for ci, cand in enumerate(candidates): for gi, gold in enumerate(goldens): if gi in matched_goldens: continue res = _judge_pair(gold, cand) - pair_results.append({"candidate_idx": ci, "golden_idx": gi, **res}) if res.get("match"): matched_goldens.add(gi) matched_candidates.add(ci) - break # candidate consumed; move to next candidate + break tp = len(matched_goldens) fp = max(0, len(candidates) - len(matched_candidates)) @@ -117,66 +137,63 @@ def judge_match(run: Run, example: Example) -> EvaluationResult: recall = tp / (tp + fn) if (tp + fn) else 0.0 f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0 - return EvaluationResult( - key="f1", - score=f1, - comment=f"P={precision:.2f} R={recall:.2f} TP={tp} FP={fp} FN={fn}", - evaluator_info={"model": JUDGE_MODEL}, - extra={ - "precision": precision, - "recall": recall, - "tp": tp, - "fp": fp, - "fn": fn, - "n_candidates": len(candidates), - "n_goldens": len(goldens), - "pairs": pair_results, - }, + _record_counts( + example.id, + {"tp": tp, "fp": fp, "fn": fn, "precision": precision, "recall": recall, "f1": f1}, ) + return { + "results": [ + {"key": "f1", "score": f1}, + {"key": "precision", "score": precision}, + {"key": "recall", "score": recall}, + {"key": "tp", "score": tp}, + {"key": "fp", "score": fp}, + {"key": "fn", "score": fn}, + {"key": "n_candidates", "score": len(candidates)}, + {"key": "n_goldens", "score": len(goldens)}, + ] + } -def aggregate_pr(runs: list[Run], examples: list[Example]) -> list[EvaluationResult]: - """Summary evaluator: micro- and macro-averaged precision/recall across the experiment.""" - micro_tp = micro_fp = micro_fn = 0 - p_macro: list[float] = [] - r_macro: list[float] = [] - for run in runs: - feedback = next( - (f for f in (run.feedback_stats or {}).values() if isinstance(f, dict)), None - ) - # Pull per-example numbers off the run's evaluator extras when available. - # We re-judge below if extras are unavailable to keep this evaluator pure. - extras = None - for ev in run.outputs and run.outputs.get("__evaluator_extras__", []) or []: - if ev.get("key") == "f1": - extras = ev.get("extra") - break - if not extras: - continue - micro_tp += extras["tp"] - micro_fp += extras["fp"] - micro_fn += extras["fn"] - p_macro.append(extras["precision"]) - r_macro.append(extras["recall"]) - del feedback # unused; reserved for future LangSmith API +def _f1(p: float, r: float) -> float: + return 2 * p * r / (p + r) if (p + r) else 0.0 - if not p_macro: - return [] + +def aggregate_pr(runs: list[Run], examples: list[Example]) -> dict[str, Any]: + """Summary evaluator: micro/macro precision-recall-F1 across the experiment. + + Reads the per-example counts that ``judge_match`` stashed in the + process-local cache. Falls back to an empty result set if the cache + is empty (e.g. summary evaluator ran in a different process). + """ + counts = _drain_counts() + if not counts: + return {"results": []} + + micro_tp = sum(int(c["tp"]) for c in counts) + micro_fp = sum(int(c["fp"]) for c in counts) + micro_fn = sum(int(c["fn"]) for c in counts) micro_p = micro_tp / (micro_tp + micro_fp) if (micro_tp + micro_fp) else 0.0 micro_r = micro_tp / (micro_tp + micro_fn) if (micro_tp + micro_fn) else 0.0 - macro_p = sum(p_macro) / len(p_macro) - macro_r = sum(r_macro) / len(r_macro) + micro_f1 = _f1(micro_p, micro_r) - def _f1(p: float, r: float) -> float: - return 2 * p * r / (p + r) if (p + r) else 0.0 + n = len(counts) + macro_p = sum(float(c["precision"]) for c in counts) / n + macro_r = sum(float(c["recall"]) for c in counts) / n + macro_f1 = sum(float(c["f1"]) for c in counts) / n - return [ - EvaluationResult(key="micro_precision", score=micro_p), - EvaluationResult(key="micro_recall", score=micro_r), - EvaluationResult(key="micro_f1", score=_f1(micro_p, micro_r)), - EvaluationResult(key="macro_precision", score=macro_p), - EvaluationResult(key="macro_recall", score=macro_r), - EvaluationResult(key="macro_f1", score=_f1(macro_p, macro_r)), - ] + return { + "results": [ + {"key": "micro_precision", "score": micro_p}, + {"key": "micro_recall", "score": micro_r}, + {"key": "micro_f1", "score": micro_f1}, + {"key": "macro_precision", "score": macro_p}, + {"key": "macro_recall", "score": macro_r}, + {"key": "macro_f1", "score": macro_f1}, + {"key": "total_tp", "score": micro_tp}, + {"key": "total_fp", "score": micro_fp}, + {"key": "total_fn", "score": micro_fn}, + ] + } diff --git a/evals/reviewer/run_eval.py b/evals/reviewer/run_eval.py index aef19606..ca026306 100644 --- a/evals/reviewer/run_eval.py +++ b/evals/reviewer/run_eval.py @@ -10,15 +10,35 @@ Usage: from __future__ import annotations import argparse +import logging +from collections.abc import Iterable from dotenv import load_dotenv -from langsmith import aevaluate +from langgraph_sdk import get_client +from langsmith import Client, aevaluate +from langsmith.schemas import Example from evals.reviewer.judge import aggregate_pr, judge_match -from evals.reviewer.target import review_pr +from evals.reviewer.target import LANGGRAPH_URL, drain_thread_ids, review_pr load_dotenv() +logger = logging.getLogger(__name__) + + +async def _cleanup_threads(thread_ids: Iterable[str]) -> None: + """Delete LangGraph threads created during the eval. + + Underlying sandboxes are reclaimed by the provider's TTL — this only + drops the LangGraph checkpoint/metadata records. + """ + sdk = get_client(url=LANGGRAPH_URL) + for tid in thread_ids: + try: + await sdk.threads.delete(tid) + except Exception as exc: + logger.warning("Failed to delete thread %s: %s", tid, exc) + async def main() -> None: ap = argparse.ArgumentParser() @@ -26,18 +46,36 @@ async def main() -> None: ap.add_argument("--experiment-prefix", default="openswe-reviewer-baseline") ap.add_argument("--max-concurrency", type=int, default=5) ap.add_argument("--limit", type=int, default=None, help="Run only the first N examples.") + ap.add_argument( + "--no-cleanup", + action="store_true", + help="Skip deleting LangGraph threads after the experiment finishes.", + ) args = ap.parse_args() - await aevaluate( - review_pr, - data=args.dataset_name, - evaluators=[judge_match], - summary_evaluators=[aggregate_pr], - experiment_prefix=args.experiment_prefix, - max_concurrency=args.max_concurrency, - num_repetitions=1, - **({"max_examples": args.limit} if args.limit else {}), - ) + data: str | list[Example] + if args.limit: + client = Client() + data = list(client.list_examples(dataset_name=args.dataset_name, limit=args.limit)) + else: + data = args.dataset_name + + try: + await aevaluate( + review_pr, + data=data, + evaluators=[judge_match], + summary_evaluators=[aggregate_pr], + experiment_prefix=args.experiment_prefix, + max_concurrency=args.max_concurrency, + num_repetitions=1, + ) + finally: + if not args.no_cleanup: + thread_ids = drain_thread_ids() + if thread_ids: + logger.info("Cleaning up %d LangGraph threads", len(thread_ids)) + await _cleanup_threads(thread_ids) if __name__ == "__main__": diff --git a/evals/reviewer/target.py b/evals/reviewer/target.py index 4a37e00b..e23615ab 100644 --- a/evals/reviewer/target.py +++ b/evals/reviewer/target.py @@ -1,15 +1,14 @@ """Target function for the reviewer eval. -Invokes the Open SWE Reviewer graph over the langgraph_sdk client and returns -the structured comments produced by the agent's `submit_review` tool call. - -The reviewer graph itself is not part of this PR — wire `REVIEWER_ASSISTANT_ID` -to whatever graph id you want to evaluate once it exists. +Spawns the reviewer graph over `langgraph_sdk` for one PR, waits for +completion, and returns every `github_comment` tool call the agent made as +the structured output for the eval. """ from __future__ import annotations import os +import threading from typing import Any from langgraph_sdk import get_client @@ -17,47 +16,77 @@ from langgraph_sdk import get_client REVIEWER_ASSISTANT_ID = os.getenv("REVIEWER_ASSISTANT_ID", "reviewer") LANGGRAPH_URL = os.getenv("LANGGRAPH_URL", "http://localhost:2024") +_THREAD_IDS: set[str] = set() +_THREAD_IDS_LOCK = threading.Lock() + + +def _record_thread_id(thread_id: str) -> None: + with _THREAD_IDS_LOCK: + _THREAD_IDS.add(thread_id) + + +def drain_thread_ids() -> set[str]: + """Return and clear thread IDs created by ``review_pr`` so far. + + Used by ``run_eval`` to delete threads after the experiment finishes. + Underlying provider sandboxes time out via their own TTL — deleting the + LangGraph thread frees the checkpoint/metadata records, not the sandbox. + """ + with _THREAD_IDS_LOCK: + snapshot = set(_THREAD_IDS) + _THREAD_IDS.clear() + return snapshot + + +def _build_user_message(inputs: dict[str, Any]) -> str: + return ( + f"Review pull request {inputs['pr_url']}.\n\n" + f"- repo: {inputs['repo']}\n" + f"- pr_number: {inputs['pr_number']}\n" + f"- title: {inputs.get('pr_title', '')}\n" + f"- base_sha: {inputs['base_sha']}\n" + f"- head_sha: {inputs['head_sha']}\n" + f"- base_ref: {inputs.get('base_ref', '')}\n" + f"- head_ref: {inputs.get('head_ref', '')}\n\n" + f"Clone the repo, check out the base SHA, fetch the PR head, and review " + f"the diff. Record each issue you find with the `github_comment` tool." + ) + async def review_pr(inputs: dict[str, Any]) -> dict[str, Any]: - """LangSmith target: run the reviewer agent on one PR. - - `inputs` carries: repo, pr_number, pr_url, base_sha, head_sha, base_ref, - head_ref, pr_title. The reviewer graph is responsible for cloning the - repo at base_sha, fetching the PR's head, and emitting structured review - comments via a `submit_review` tool whose args become the graph output. - - Returns: {"comments": [{file, line, severity, body}, ...]}. - """ + """LangSmith target: run the reviewer agent on one PR.""" client = get_client(url=LANGGRAPH_URL) thread = await client.threads.create() + thread_id: str = thread["thread_id"] + _record_thread_id(thread_id) result = await client.runs.wait( - thread["thread_id"], + thread_id, assistant_id=REVIEWER_ASSISTANT_ID, - input={"pr": inputs}, + input={"messages": [{"role": "user", "content": _build_user_message(inputs)}]}, + config={"configurable": {"__is_for_execution__": True}}, ) return {"comments": _extract_comments(result)} -def _extract_comments(result: Any) -> list[dict]: - """Pull the submit_review payload out of the graph's final state. - - Supports two shapes: - 1. Graph state contains a top-level `review` field populated by the tool - (preferred — wire the reviewer graph to set this). - 2. Last AI message includes a `submit_review` tool call; we parse args. - """ - if isinstance(result, dict): - if isinstance(result.get("review"), dict) and "comments" in result["review"]: - return list(result["review"]["comments"]) - if isinstance(result.get("comments"), list): - return list(result["comments"]) - - messages = result.get("messages") or [] - for msg in reversed(messages): - tool_calls = msg.get("tool_calls") if isinstance(msg, dict) else None - for tc in tool_calls or []: - if tc.get("name") == "submit_review": - args = tc.get("args") or {} - if isinstance(args.get("comments"), list): - return list(args["comments"]) - return [] +def _extract_comments(result: Any) -> list[dict[str, Any]]: + """Collect every `github_comment` tool call from the run's message stream.""" + if not isinstance(result, dict): + return [] + comments: list[dict[str, Any]] = [] + for msg in result.get("messages") or []: + if not isinstance(msg, dict): + continue + for tc in msg.get("tool_calls") or []: + if tc.get("name") != "github_comment": + continue + args = tc.get("args") or {} + if {"file", "line", "body", "severity"} <= args.keys(): + comments.append( + { + "file": args["file"], + "line": args["line"], + "body": args["body"], + "severity": args["severity"], + } + ) + return comments diff --git a/langgraph.json b/langgraph.json index 57ae6626..1210bbb6 100644 --- a/langgraph.json +++ b/langgraph.json @@ -2,7 +2,8 @@ "$schema": "https://langgra.ph/schema.json", "python_version": "3.12", "graphs": { - "agent": "agent.server:get_agent" + "agent": "agent.server:get_agent", + "reviewer": "agent.reviewer:get_reviewer_agent" }, "dependencies": ["."], "http": {