admin/OpenJarvis
1
0
Fork 0
mirror of https://github.com/open-jarvis/OpenJarvis.git synced 2026-07-29 18:40:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

hybrid: wire Advisors + ToolOrchestra workers through mini-SWE-agent

Browse Source 
Same 'swe_use_agent_loop' opt-in flag as Conductor/SkillOrchestra.

Advisors (swe mode):
- Initial executor pass = full mini-SWE-agent run on a fresh workdir.
- Advisor (local) critiques the produced patch (sees summary + diff).
- Final executor pass = SECOND mini-SWE-agent run on ANOTHER fresh
  workdir, with the initial summary + advisor feedback folded into the
  prompt. Don't reuse the initial workdir — that would smuggle the
  bad-fix into the new attempt; the final pass should incorporate only
  the parts of the advisor feedback that hold up.

ToolOrchestra (swe mode):
- One SHARED workdir created at start of run, cloned from the SWE-bench
  repo at base_commit.
- Each call_worker action whose worker is a solver (vllm/anthropic)
  runs as a mini-SWE-agent subloop on the shared workdir. Web-search
  workers stay one-shot (search isn't an agent loop). OpenAI workers
  fall back to one-shot too (loop tool format isn't wired for OpenAI).
- After loop ends, the framed final answer wraps the working-tree
  git diff so the swebench scorer picks it up.
This commit is contained in:
Andrew Park
2026-05-15 13:42:10 -07:00
parent e71342b502
commit d83ae02c6b
2 changed files with 234 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openjarvis/agents/hybrid/advisors.py
+124
View File
@@ -27,8 +27,17 @@ import json
import urllib.request
from typing import Any, Dict, Optional, Tuple
import shutil
import tempfile
from pathlib import Path
from openjarvis.agents._stubs import AgentContext
from openjarvis.agents.hybrid._base import LocalCloudAgent
from openjarvis.agents.hybrid.mini_swe_agent import (
_clone_repo,
_extract_diff,
run_swe_agent_loop,
)
from openjarvis.core.registry import AgentRegistry
@@ -96,6 +105,16 @@ class AdvisorsAgent(LocalCloudAgent):
) -> Tuple[str, Dict[str, Any]]:
question = input
cfg = self._cfg
task_meta = (context.metadata.get("task") if context is not None else {}) or {}
swe_mode = (
bool(cfg.get("swe_use_agent_loop"))
and bool(task_meta.get("problem_statement"))
and bool(task_meta.get("repo"))
and bool(task_meta.get("base_commit"))
)
if swe_mode:
return self._run_swe(question, task_meta, cfg)
executor_max_tokens = int(cfg.get("executor_max_tokens", 4096))
advisor_max_tokens = int(cfg.get("advisor_max_tokens", 1024))
advisor_temperature = float(cfg.get("advisor_temperature", 0.2))
@@ -159,5 +178,110 @@ class AdvisorsAgent(LocalCloudAgent):
}
return final_answer, meta
# ------------------------------------------------------------------
# SWE-bench variant: each executor pass is a full mini-SWE-agent run.
# ------------------------------------------------------------------
def _run_swe(
self,
question: str,
task: Dict[str, Any],
cfg: Dict[str, Any],
) -> Tuple[str, Dict[str, Any]]:
if not self._local_endpoint or not self._local_model:
raise ValueError(
"AdvisorsAgent (swe mode) still needs local_model + local_endpoint "
"for the advisor critique step."
)
# Each executor pass is its own mini-SWE-agent run with a FRESH
# workdir — the advisor doesn't get a workdir, just the patch text.
max_turns = int(cfg.get("swe_max_turns", 30))
bash_timeout = int(cfg.get("swe_bash_timeout_s", 120))
output_cap = int(cfg.get("swe_output_cap", 10_000))
turn_max_tokens = int(cfg.get("swe_turn_max_tokens", 4096))
# 1. Initial executor pass
initial_out = run_swe_agent_loop(
task,
backbone="cloud",
backbone_model=self._cloud_model,
cloud_endpoint=self._cloud_endpoint,
initial_prompt=question,
max_turns=max_turns,
bash_timeout=bash_timeout,
output_cap=output_cap,
turn_max_tokens=turn_max_tokens,
trace_prefix="advisors_executor1",
)
# 2. Advisor pass — local model critiques the produced patch
local_model = _resolve_local_model(self._local_endpoint, self._local_model)
advisor_prompt = ADVISOR_TEMPLATE.format(
question=question,
initial_response=(
f"Summary: {initial_out['final_summary']}\n\n"
f"Patch produced:\n```diff\n{initial_out['patch']}```"
),
)
advisor_text, adv_in, adv_out = self._call_vllm(
local_model,
self._local_endpoint,
user=advisor_prompt,
max_tokens=int(cfg.get("advisor_max_tokens", 2048)),
temperature=float(cfg.get("advisor_temperature", 0.2)),
enable_thinking=False,
)
# 3. Final executor pass — FRESH workdir, advisor feedback folded
# into the initial prompt. (Don't reuse initial workdir — that
# would smuggle the bad-fix into the new attempt; we want the
# final pass to incorporate ONLY the parts of the advisor's
# feedback that hold up.)
final_prompt = (
f"{question}\n\n"
f"-----\n"
f"You previously attempted a fix. Your initial summary was:\n"
f"{initial_out['final_summary']}\n\n"
f"An advisor reviewed your attempt and wrote this feedback:\n"
f"{advisor_text}\n\n"
f"Incorporate the advisor's feedback where it improves correctness; "
f"ignore where it is wrong. Produce your best fix now."
)
final_out = run_swe_agent_loop(
task,
backbone="cloud",
backbone_model=self._cloud_model,
cloud_endpoint=self._cloud_endpoint,
initial_prompt=final_prompt,
max_turns=max_turns,
bash_timeout=bash_timeout,
output_cap=output_cap,
turn_max_tokens=turn_max_tokens,
trace_prefix="advisors_executor2",
)
tokens_local = adv_in + adv_out
tokens_cloud = (
initial_out["tokens_in"] + initial_out["tokens_out"]
+ final_out["tokens_in"] + final_out["tokens_out"]
)
cost = initial_out["cost_usd"] + final_out["cost_usd"]
meta: Dict[str, Any] = {
"tokens_local": tokens_local,
"tokens_cloud": tokens_cloud,
"cost_usd": cost,
"turns": initial_out["turns"] + 1 + final_out["turns"],
"traces": {
"swe_mode": True,
"initial_summary": initial_out["final_summary"],
"initial_patch_chars": len(initial_out["patch"]),
"advisor_feedback": advisor_text,
"final_summary": final_out["final_summary"],
"final_patch_chars": len(final_out["patch"]),
},
}
return final_out["answer"], meta
__all__ = ["AdvisorsAgent"]
src/openjarvis/agents/hybrid/toolorchestra.py
+110 -6
View File
@@ -43,6 +43,10 @@ import json
import re
from typing import Any, Dict, List, Optional, Tuple
import shutil
import tempfile
from pathlib import Path
from openjarvis.agents._stubs import AgentContext
from openjarvis.agents.hybrid._base import (
ANTHROPIC_WEB_SEARCH_TOOL,
@@ -53,6 +57,11 @@ from openjarvis.agents.hybrid._prices import (
is_gpt5_family,
supports_temperature,
)
from openjarvis.agents.hybrid.mini_swe_agent import (
_clone_repo,
_extract_diff,
run_swe_agent_loop,
)
from openjarvis.core.registry import AgentRegistry
@@ -263,6 +272,52 @@ def _call_worker(
raise ValueError(f"unsupported worker type: {wtype!r}")
def _swe_call_worker(
worker: Dict[str, Any],
prompt: str,
cfg: Dict[str, Any],
task: Dict[str, Any],
workdir: Path,
turn: int,
) -> Tuple[str, int, int, bool, float, int]:
"""SWE-bench worker dispatch: route solver workers through
run_swe_agent_loop on a shared workdir. Web-search workers fall back
to the regular one-shot dispatch (search isn't an agent loop)."""
wtype = worker.get("type", "openai")
if wtype == "anthropic-web-search":
# Search workers stay one-shot.
return _call_worker(worker, prompt, cfg)
if wtype == "vllm":
backbone = "local"
endpoint = worker.get("base_url")
elif wtype == "anthropic":
backbone = "cloud"
endpoint = None
else:
# OpenAI workers fall back to one-shot.
return _call_worker(worker, prompt, cfg)
out = run_swe_agent_loop(
task,
backbone=backbone,
backbone_model=worker["model"],
cloud_endpoint="anthropic",
local_endpoint=endpoint,
initial_prompt=prompt,
max_turns=int(cfg.get("swe_max_turns", 30)),
bash_timeout=int(cfg.get("swe_bash_timeout_s", 120)),
output_cap=int(cfg.get("swe_output_cap", 10_000)),
turn_max_tokens=int(cfg.get("swe_turn_max_tokens", 4096)),
trace_prefix=f"toolorch_turn{turn}",
workdir=workdir,
)
is_local = backbone == "local"
return (
out["final_summary"] or out["answer"],
out["tokens_in"], out["tokens_out"],
is_local, 0.0, 0,
)
@AgentRegistry.register("toolorchestra")
class ToolOrchestraAgent(LocalCloudAgent):
"""Prompted multi-turn dispatcher over a mixed worker pool.
@@ -290,6 +345,30 @@ class ToolOrchestraAgent(LocalCloudAgent):
max_turns = int(cfg.get("max_turns", 6))
orch_max_tokens = int(cfg.get("orchestrator_max_tokens", 1024))
task_meta = (context.metadata.get("task") if context is not None else {}) or {}
swe_mode = (
bool(cfg.get("swe_use_agent_loop"))
and bool(task_meta.get("problem_statement"))
and bool(task_meta.get("repo"))
and bool(task_meta.get("base_commit"))
)
shared_workdir: Optional[Path] = None
if swe_mode:
shared_workdir = Path(tempfile.mkdtemp(
prefix=f"toolorch-swe-{task_meta.get('task_id','x')}-"
))
try:
_clone_repo(task_meta["repo"], task_meta["base_commit"], shared_workdir)
except Exception:
shutil.rmtree(shared_workdir, ignore_errors=True)
raise
self.record_trace_event({
"kind": "toolorchestra_swe_workdir",
"workdir": str(shared_workdir),
"repo": task_meta["repo"],
"base_commit": task_meta["base_commit"],
})
history: List[Dict[str, Any]] = []
tokens_local = 0
tokens_cloud = 0
@@ -346,9 +425,17 @@ class ToolOrchestraAgent(LocalCloudAgent):
break
continue
worker = workers[wid]
w_text, w_in, w_out, is_local, extra_cost, n_searches = _call_worker(
worker, str(w_input), cfg
)
if swe_mode and shared_workdir is not None:
w_text, w_in, w_out, is_local, extra_cost, n_searches = (
_swe_call_worker(
worker, str(w_input), cfg, task_meta,
shared_workdir, turn,
)
)
else:
w_text, w_in, w_out, is_local, extra_cost, n_searches = (
_call_worker(worker, str(w_input), cfg)
)
if is_local:
tokens_local += w_in + w_out
else:
@@ -372,9 +459,15 @@ class ToolOrchestraAgent(LocalCloudAgent):
if final_answer is None:
# Hard fallback: call the strongest worker (last) directly.
worker = workers[-1]
ans, w_in, w_out, is_local, extra_cost, _ = _call_worker(
worker, question, cfg
)
if swe_mode and shared_workdir is not None:
ans, w_in, w_out, is_local, extra_cost, _ = _swe_call_worker(
worker, question, cfg, task_meta,
shared_workdir, max_turns + 1,
)
else:
ans, w_in, w_out, is_local, extra_cost, _ = _call_worker(
worker, question, cfg
)
if is_local:
tokens_local += w_in + w_out
else:
@@ -393,6 +486,17 @@ class ToolOrchestraAgent(LocalCloudAgent):
})
final_answer = ans
# In SWE mode, the authoritative output is the working-tree diff —
# frame it (the runner extracts it via the scorer's ```diff fence).
if swe_mode and shared_workdir is not None:
patch = _extract_diff(shared_workdir)
if patch.strip():
final_answer = (
f"{final_answer}\n\n```diff\n{patch}```"
if final_answer else f"```diff\n{patch}```"
)
shutil.rmtree(shared_workdir, ignore_errors=True)
meta = {
"tokens_local": tokens_local,
"tokens_cloud": tokens_cloud,