Hoan/GenericAgent
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(conductor): manage model selection in agent memory only

Browse Source
This commit is contained in:
Liang Jiaqing committed 2026-10-15 21:34:28 +08:00
1 parent e4e610d6a1
commit 24f7d6a0dc
2 files changed
+167 -265
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

frontends/conductor.py
+43 -152
View File
@@ -1,6 +1,5 @@
import os, sys, re, time, json, uuid, queue, asyncio, threading, argparse, base64, secrets
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, Any, Optional, List
from contextlib import asynccontextmanager, suppress
@@ -11,7 +10,6 @@ from pydantic import BaseModel
FRONTENDS_DIR = os.path.dirname(os.path.abspath(__file__))
if FRONTENDS_DIR not in sys.path: sys.path.insert(0, FRONTENDS_DIR)
from desktop_settings import DesktopSettingsError, update_settings
def _resolve_ga_root() -> str:
"""Use the external core selected by the package-owned desktop bridge when valid."""
@@ -72,41 +70,6 @@ HOST = args.host
PORT = args.port
SETTINGS_PATH = Path.home() / ".ga_desktop_settings.json"
def _settings_doc() -> dict:
try:
doc = json.loads(SETTINGS_PATH.read_text(encoding="utf-8"))
return doc if isinstance(doc, dict) else {}
except Exception:
return {}
def _persist_conductor_llm_no(llm_no: int) -> None:
def mutate(document: dict) -> None:
section = document.get("conductor")
if not isinstance(section, dict):
section = {}
document["conductor"] = section
section["llmNo"] = llm_no
update_settings(SETTINGS_PATH, mutate)
def _conductor_llm_no() -> Optional[int]:
"""Read the model index bound to the conductor session.
Falls back to the legacy desktop default ui.llmNo for existing installs."""
doc = _settings_doc()
for section in (doc.get("conductor"), doc.get("ui")):
if isinstance(section, dict) and section.get("llmNo") is not None:
try:
return int(section.get("llmNo"))
except (TypeError, ValueError):
pass
return None
def _client_usable(agent: "GenericAgent") -> bool:
return hasattr(getattr(agent, "llmclient", None), "backend")
@@ -131,89 +94,12 @@ def _activate_model(agent: "GenericAgent", no: int) -> None:
agent.next_llm(no)
def _runtime_model_state(agent: "GenericAgent", configured: Optional[int], reason: Optional[str]) -> dict:
effective = getattr(agent, "llm_no", None) if _client_usable(agent) else None
current = None
if _client_usable(agent):
with suppress(Exception):
current = str(agent.llmclient.backend.name)
return {"configured": configured, "effective": effective,
"fallbackReason": reason, "current": current}
def _apply_desktop_model(agent: "GenericAgent") -> dict:
"""Make the conductor's session reflect the current desktop config before a task:
switch to its bound model if one is set, otherwise still refresh sessions from
mykey so live key/model edits (e.g. importing keys) take effect without a restart.
next_llm() already reloads internally; the no-bound-model branch must reload too,
or a conductor started on an empty/stale mykey would never pick up imported keys."""
doc = _settings_doc()
conductor_cfg = doc.get("conductor") if isinstance(doc.get("conductor"), dict) else {}
ui_cfg = doc.get("ui") if isinstance(doc.get("ui"), dict) else {}
raw_configured = conductor_cfg.get("llmNo")
configured = _parse_model_no(raw_configured)
ui_default = _parse_model_no(ui_cfg.get("llmNo"))
try:
agent.load_llm_sessions() # mtime-guarded; rebuilds only when mykey changed
except Exception as e:
print(f"[conductor] failed to refresh model sessions: {e}", file=sys.stderr)
clients = getattr(agent, "llmclients", []) or []
configured_failed = False
if _usable_model(agent, configured):
try:
_activate_model(agent, configured)
return _runtime_model_state(agent, configured, None)
except Exception as e:
configured_failed = True
print(f"[conductor] configured model #{configured} is unavailable: {e}", file=sys.stderr)
if _usable_model(agent, ui_default):
if raw_configured is None:
reason = "ui_default"
elif configured_failed:
reason = "configured_unavailable"
elif configured is not None:
reason = "configured_unavailable" if 0 <= configured < len(clients) else "invalid_configured"
else:
reason = "invalid_configured"
try:
_activate_model(agent, ui_default)
return _runtime_model_state(agent, configured, reason)
except Exception as e:
print(f"[conductor] UI default model #{ui_default} is unavailable: {e}", file=sys.stderr)
for i, client in enumerate(clients):
if hasattr(client, "backend"):
try:
_activate_model(agent, i)
return _runtime_model_state(agent, configured, "first_available")
except Exception:
continue
print("[conductor] no usable model is available", file=sys.stderr)
return {"configured": configured, "effective": None,
"fallbackReason": "no_models", "current": None}
def _selected_conductor_llm_no(agent: "GenericAgent") -> int:
configured = _conductor_llm_no()
return configured if _usable_model(agent, configured) else agent.llm_no
def _set_conductor_llm_no(agent: "GenericAgent", value: Any) -> int:
"""Persist the standalone UI binding without mutating an in-flight client.
The Conductor loop applies this binding at its existing idle task boundary, so
a selection made immediately before sending a message controls that next turn.
"""
no = _parse_model_no(value)
if no is None or not _usable_model(agent, no):
raise ValueError(f"llm index out of range or unavailable: {value}")
_persist_conductor_llm_no(no)
return no
def _runtime_model_state(agent: "GenericAgent") -> dict:
usable = _client_usable(agent)
return {"configured": None,
"effective": agent.llm_no if usable else None,
"fallbackReason": None if usable else "no_models",
"current": str(agent.llmclient.backend.name) if usable else None}
def _select_llm(agent: "GenericAgent", llm: Any) -> bool:
@@ -447,7 +333,7 @@ class SubagentPool:
agent.inc_out = True
agent.verbose = False
agent.no_print = True
if not _select_llm(agent, llm): _apply_desktop_model(agent)
_select_llm(agent, llm)
th = start_agent_runner(agent, f"subagent-{sid}")
state = SubAgentState(id=sid, agent=agent, prompt=prompt, status="running", thread=th)
with self.lock: self.subagents[sid] = state
@@ -532,24 +418,30 @@ class Conductor:
self._runner_thread: Optional[threading.Thread] = None
self.log: list = []
self._model_lock = threading.RLock()
self._model_state = {
"configured": None,
"effective": None,
"fallbackReason": "no_models",
"current": None,
"running": False,
}
self._model_running = False
def model_snapshot(self) -> dict:
with self._model_lock:
return dict(self._model_state)
return {**_runtime_model_state(self.agent), "running": self._model_running}
def _publish_model_state(self, state: dict, running: bool) -> None:
snapshot = {**state, "running": bool(running)}
with self._model_lock:
self._model_state = snapshot
self._model_running = bool(running)
schedule_broadcast({"type": "model", "model": snapshot})
def select_model(self, value: Any) -> int:
with self._model_lock:
if self._model_running or self.agent.task_queue.unfinished_tasks:
raise ValueError("conductor is busy; switch models after the task finishes")
no = _parse_model_no(value)
if no is None:
raise ValueError(f"invalid model index: {value}")
self.agent.load_llm_sessions()
_activate_model(self.agent, no)
self._publish_model_state(_runtime_model_state(self.agent), running=False)
return self.agent.llm_no
def notify(self, event: dict): self.inbox.put(event)
def _wait_for_agent_idle(self, timeout: float = 10.0) -> None:
@@ -657,24 +549,23 @@ API: {base}；requests，GET /readme查用法，GET /chat读未读对话，GET /
# mutate the live client until that previous task has reached
# ``task_done``; the queue join is the authoritative idle barrier.
self._wait_for_agent_idle()
# Settings can change while this long-lived process remains healthy.
# Refresh only at the task boundary so an in-flight task keeps its
# client, while the next task observes the newly persisted binding.
model_state = _apply_desktop_model(self.agent)
if model_state.get("effective") is None:
self._publish_model_state(model_state, running=False)
self._record_unavailable_model(events)
deferred_events = events
continue
kind = events[0].get("type")
other_events = [event for event in events if event.get("type") != kind]
events = [event for event in events if event.get("type") == kind]
for event in other_events:
self.inbox.put(event)
prompt = self._build_prompt(events)
self._publish_model_state(model_state, running=True)
try:
with self._model_lock:
self.agent.load_llm_sessions()
model_state = _runtime_model_state(self.agent)
if model_state.get("effective") is None:
self._publish_model_state(model_state, running=False)
self._record_unavailable_model(events)
deferred_events = events
continue
kind = events[0].get("type")
other_events = [event for event in events if event.get("type") != kind]
events = [event for event in events if event.get("type") == kind]
for event in other_events:
self.inbox.put(event)
prompt = self._build_prompt(events)
dq = self.agent.put_task(prompt, source="conductor")
self._publish_model_state(model_state, running=True)
try:
self._drain(dq, events)
finally:
self._publish_model_state(model_state, running=False)
@@ -682,7 +573,7 @@ API: {base}；requests，GET /readme查用法，GET /chat读未读对话，GET /
def start(self):
self.agent = GenericAgent()
self._publish_model_state(_apply_desktop_model(self.agent), running=False)
self._publish_model_state(_runtime_model_state(self.agent), running=False)
threading.Thread(target=self._run, name="conductor-loop", daemon=True).start()
@@ -838,17 +729,17 @@ async def websocket(ws: WebSocket):
"log": conductor.log, "running": running,
"model": conductor.model_snapshot(),
"llms": conductor.agent.list_llms(),
"llm": _selected_conductor_llm_no(conductor.agent)})
"llm": conductor.agent.llm_no})
while True:
data = await ws.receive_json()
if "llm" in data:
try:
llm_no = _set_conductor_llm_no(conductor.agent, data["llm"])
except (TypeError, ValueError, OSError, DesktopSettingsError) as error:
llm_no = conductor.select_model(data["llm"])
except (TypeError, ValueError, RuntimeError) as error:
await ws.send_json({
"type": "model_error",
"error": str(error),
"llm": _selected_conductor_llm_no(conductor.agent),
"llm": conductor.agent.llm_no,
})
else:
await broadcast({"type": "model_selected", "llm": llm_no})
frontends/tests/test_conductor_model_routing.py
+124 -113
View File
@@ -61,6 +61,7 @@ class FakeAgent:
self.next_llm_calls: list[int] = []
self.reloads = 0
self.failing = failing or set()
self.task_queue = queue.Queue()
def load_llm_sessions(self):
self.reloads += 1
@@ -79,20 +80,6 @@ class FakeAgent:
return client.backend.model if model else client.backend.name
def test_out_of_range_config_falls_back_without_modulo(monkeypatch):
agent = FakeAgent(["zero", "one", "two", "three"])
monkeypatch.setattr(
conductor,
"_settings_doc",
lambda: {"conductor": {"llmNo": 99}, "ui": {"llmNo": 1}},
)
state = conductor._apply_desktop_model(agent)
assert agent.next_llm_calls == [1]
assert state["effective"] == 1
assert state["fallbackReason"] == "invalid_configured"
def test_conductor_default_port_override_is_e2e_scoped(monkeypatch):
monkeypatch.delenv(conductor.E2E_REPORT_DIR_ENV, raising=False)
@@ -114,45 +101,7 @@ def test_conductor_port_parser_rejects_invalid_values(value):
conductor._parse_conductor_port(value)
def test_unusable_configured_client_falls_back_to_ui_default(monkeypatch):
agent = FakeAgent(["zero", "one", None, "three"])
monkeypatch.setattr(
conductor,
"_settings_doc",
lambda: {"conductor": {"llmNo": 2}, "ui": {"llmNo": 1}},
)
state = conductor._apply_desktop_model(agent)
assert agent.next_llm_calls == [1]
assert state["effective"] == 1
assert state["fallbackReason"] == "configured_unavailable"
def test_configured_activation_failure_falls_back_to_ui_default(monkeypatch):
agent = FakeAgent(["zero", "one", "two"], failing={2})
monkeypatch.setattr(
conductor,
"_settings_doc",
lambda: {"conductor": {"llmNo": 2}, "ui": {"llmNo": 1}},
)
state = conductor._apply_desktop_model(agent)
assert agent.next_llm_calls == [2, 1]
assert state["effective"] == 1
assert state["fallbackReason"] == "configured_unavailable"
def test_missing_config_uses_first_usable_when_ui_is_invalid(monkeypatch):
agent = FakeAgent([None, None, "two", "three"])
monkeypatch.setattr(conductor, "_settings_doc", lambda: {"ui": {"llmNo": 99}})
state = conductor._apply_desktop_model(agent)
assert agent.next_llm_calls == [2]
assert state["effective"] == 2
assert state["fallbackReason"] == "first_available"
def test_explicit_numeric_worker_model_rejects_out_of_range():
@@ -164,52 +113,12 @@ def test_explicit_numeric_worker_model_rejects_out_of_range():
assert agent.next_llm_calls == []
def test_standalone_model_selection_persists_without_mutating_live_client(tmp_path, monkeypatch):
settings_path = tmp_path / "settings.json"
settings_path.write_text('{"theme":"dark","ui":{"llmNo":0}}', encoding="utf-8")
monkeypatch.setattr(conductor, "SETTINGS_PATH", settings_path)
agent = FakeAgent(["zero", "one", "two"])
selected = conductor._set_conductor_llm_no(agent, 2)
assert selected == 2
assert agent.llm_no == 0
assert agent.next_llm_calls == []
assert conductor._settings_doc() == {
"theme": "dark",
"ui": {"llmNo": 0},
"conductor": {"llmNo": 2},
}
def test_standalone_model_selection_rejects_unavailable_model(tmp_path, monkeypatch):
settings_path = tmp_path / "settings.json"
settings_path.write_text('{"theme":"dark"}', encoding="utf-8")
monkeypatch.setattr(conductor, "SETTINGS_PATH", settings_path)
agent = FakeAgent(["zero", None])
with pytest.raises(ValueError, match="unavailable"):
conductor._set_conductor_llm_no(agent, 1)
assert conductor._settings_doc() == {"theme": "dark"}
def test_standalone_model_selection_applies_at_next_task_boundary(tmp_path, monkeypatch):
settings_path = tmp_path / "settings.json"
monkeypatch.setattr(conductor, "SETTINGS_PATH", settings_path)
agent = FakeAgent(["zero", "one", "two"])
conductor._set_conductor_llm_no(agent, 2)
state = conductor._apply_desktop_model(agent)
assert agent.next_llm_calls == [2]
assert state["configured"] == 2
assert state["effective"] == 2
assert state["fallbackReason"] is None
def test_runtime_model_snapshot_is_broadcast_with_running_state(monkeypatch):
instance = conductor.Conductor()
instance.agent = FakeAgent(["model-one"])
payloads: list[dict] = []
monkeypatch.setattr(conductor, "schedule_broadcast", payloads.append)
state = {
@@ -221,12 +130,12 @@ def test_runtime_model_snapshot_is_broadcast_with_running_state(monkeypatch):
instance._publish_model_state(state, running=True)
assert instance.model_snapshot() == {**state, "running": True}
assert instance.model_snapshot() == {**conductor._runtime_model_state(instance.agent), "running": True}
assert payloads == [{"type": "model", "model": {**state, "running": True}}]
def test_long_lived_conductor_refreshes_model_between_tasks(monkeypatch):
"""A saved binding waits for task cleanup, then affects the next task."""
"""Display completion does not permit switching before task cleanup."""
class EndOfInbox(Exception):
pass
@@ -254,7 +163,6 @@ def test_long_lived_conductor_refreshes_model_between_tasks(monkeypatch):
# Keep each event as a separate task boundary for this regression.
return True
selected = {"llmNo": 1}
timeline: list[tuple[str, int]] = []
allow_first_cleanup = threading.Event()
first_cleanup_done = threading.Event()
@@ -276,10 +184,7 @@ def test_long_lived_conductor_refreshes_model_between_tasks(monkeypatch):
self.task_models.append(self.llm_no)
timeline.append(("submit", self.llm_no))
if len(self.task_models) == 1:
# Simulate save_config while task one is already submitted. The
# display queue reports done before the agent runner persists its
# final state and calls task_done().
selected["llmNo"] = 2
# Display completion precedes task cleanup.
assert self.llm_no == 1
self.task_queue.put("first")
@@ -302,13 +207,9 @@ def test_long_lived_conductor_refreshes_model_between_tasks(monkeypatch):
instance = conductor.Conductor()
instance.agent = TaskAgent()
instance.agent.next_llm(1)
instance.inbox = BoundaryInbox()
payloads: list[dict] = []
monkeypatch.setattr(
conductor,
"_settings_doc",
lambda: {"conductor": {"llmNo": selected["llmNo"]}},
)
monkeypatch.setattr(conductor, "schedule_broadcast", payloads.append)
monkeypatch.setattr(conductor, "start_agent_runner", lambda *_args: None)
monkeypatch.setattr(conductor.time, "sleep", lambda _seconds: None)
@@ -333,6 +234,8 @@ def test_long_lived_conductor_refreshes_model_between_tasks(monkeypatch):
assert instance.agent.task_models == [1]
assert timeline == [("select", 1), ("submit", 1)]
with pytest.raises(ValueError, match="busy"):
instance.select_model(2)
allow_first_cleanup.set()
assert first_cleanup_done.wait(timeout=2)
runner.join(timeout=2)
@@ -342,21 +245,20 @@ def test_long_lived_conductor_refreshes_model_between_tasks(monkeypatch):
assert not runner.is_alive()
assert run_errors == []
assert instance.agent.task_models == [1, 2]
assert instance.agent.task_models == [1, 1]
assert timeline == [
("select", 1),
("submit", 1),
("select", 2),
("submit", 2),
("submit", 1),
]
assert instance.model_snapshot() == {
"configured": 2,
"effective": 2,
"configured": None,
"effective": 1,
"fallbackReason": None,
"current": "two",
"current": "one",
"running": False,
}
assert [item["model"]["effective"] for item in payloads] == [1, 1, 2, 2]
assert [item["model"]["effective"] for item in payloads] == [1, 1, 1, 1]
assert [item["model"]["running"] for item in payloads] == [True, False, True, False]
@@ -397,6 +299,7 @@ def test_conductor_defers_events_without_models_and_recovers(monkeypatch):
def load_llm_sessions(self):
if ready["value"] and not self.llmclients:
self.llmclients = [Client("recovered")]
self.llmclient = self.llmclients[0]
def next_llm(self, no: int):
self.llm_no = no
@@ -417,7 +320,6 @@ def test_conductor_defers_events_without_models_and_recovers(monkeypatch):
instance.agent = RecoveringAgent()
instance.inbox = BoundaryInbox()
payloads: list[dict] = []
monkeypatch.setattr(conductor, "_settings_doc", lambda: {})
monkeypatch.setattr(conductor, "schedule_broadcast", payloads.append)
monkeypatch.setattr(conductor, "start_agent_runner", lambda *_args: None)
monkeypatch.setattr(conductor.time, "sleep", lambda _seconds: None)
@@ -453,3 +355,112 @@ def test_parallel_worker_model_selection_keeps_agents_isolated():
assert selected == [True] * 10
assert [agent.llm_no for agent in agents] == requested
assert [agent.next_llm_calls for agent in agents] == [[no] for no in requested]
def test_model_selection_is_memory_only_and_survives_refresh(monkeypatch):
instance = conductor.Conductor()
instance.agent = FakeAgent(["zero", "one", "two"])
monkeypatch.setattr(conductor, "schedule_broadcast", lambda _: None)
def forbidden(*args, **kwargs):
raise AssertionError("model selection must not access settings files")
monkeypatch.setattr("builtins.open", forbidden)
monkeypatch.setattr(Path, "open", forbidden)
assert instance.select_model(2) == 2
instance.agent.load_llm_sessions()
assert instance.model_snapshot()["effective"] == 2
assert instance.agent.next_llm_calls == [2]
# A fresh agent uses its own default, not a previously saved selection.
other = conductor.Conductor()
other.agent = FakeAgent(["zero", "one", "two"])
assert other.model_snapshot()["effective"] == 0
@pytest.mark.parametrize("value", [99, -1, True, None, "invalid", 1])
def test_invalid_model_selection_keeps_current_agent(value, monkeypatch):
instance = conductor.Conductor()
instance.agent = FakeAgent(["zero", None])
monkeypatch.setattr(conductor, "schedule_broadcast", lambda _: None)
with pytest.raises(ValueError):
instance.select_model(value)
assert instance.agent.llm_no == 0
assert instance.agent.next_llm_calls == []
def test_running_model_selection_is_rejected(monkeypatch):
instance = conductor.Conductor()
instance.agent = FakeAgent(["zero", "one"])
instance._model_running = True
with pytest.raises(ValueError, match="busy"):
instance.select_model(1)
assert instance.agent.next_llm_calls == []
def test_actual_generic_agent_next_llm_preserves_history(monkeypatch):
# Execute the real method with inert clients; no credentials or LLM calls.
import ast
tree = ast.parse((ROOT / "agentmain.py").read_text(encoding="utf-8"))
cls = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == "GenericAgent")
method = next(n for n in cls.body if isinstance(n, ast.FunctionDef) and n.name == "next_llm")
namespace = {"load_tool_schema": lambda: None}
exec(compile(ast.Module(body=[method], type_ignores=[]), "agentmain.py", "exec"), namespace)
instance = conductor.Conductor()
instance.agent = FakeAgent(["zero", "one"])
instance.agent.llmclient.backend.history = [{"role": "user", "content": "history"}]
history = instance.agent.llmclient.backend.history
instance.agent.next_llm = types.MethodType(namespace["next_llm"], instance.agent)
monkeypatch.setattr(conductor, "schedule_broadcast", lambda _: None)
assert instance.select_model(1) == 1
assert instance.agent.llmclient is instance.agent.llmclients[1]
assert instance.agent.llmclient.backend.history is history
assert instance.agent.llmclient.last_tools == ""
def test_websocket_switch_reconnect_and_busy_error(monkeypatch):
from fastapi.testclient import TestClient
instance = conductor.Conductor()
instance.agent = FakeAgent(["zero", "one"])
instance.agent.list_llms = lambda: [(i, c.backend.name, i == instance.agent.llm_no)
for i, c in enumerate(instance.agent.llmclients)]
monkeypatch.setattr(conductor, "conductor", instance)
monkeypatch.setattr(conductor, "schedule_broadcast", lambda _: None)
# Do not enter TestClient lifespan: no real runner or IM poller is started.
client = TestClient(conductor.app)
with client.websocket_connect("/ws") as ws:
assert ws.receive_json()["llm"] == 0
ws.send_json({"llm": 1})
assert ws.receive_json() == {"type": "model_selected", "llm": 1}
instance.agent.task_queue.put("running")
ws.send_json({"llm": 0})
error = ws.receive_json()
assert error["type"] == "model_error"
assert error["llm"] == 1
instance.agent.task_queue.get_nowait()
instance.agent.task_queue.task_done()
ws.send_json({"llm": 99})
assert ws.receive_json()["type"] == "model_error"
with client.websocket_connect("/ws") as ws:
hello = ws.receive_json()
assert hello["llm"] == hello["model"]["effective"] == 1
def test_conductor_has_no_desktop_settings_dependency():
source = CONDUCTOR_PATH.read_text(encoding="utf-8")
for removed in ["desktop_settings", "SETTINGS_PATH", "_apply_desktop_model", "ui_default"]:
assert removed not in source
def test_start_uses_generic_agent_default_without_files(monkeypatch):
agent = FakeAgent(["zero", "one"])
agent.next_llm(1)
monkeypatch.setattr(conductor, "GenericAgent", lambda: agent)
monkeypatch.setattr(conductor, "schedule_broadcast", lambda _: None)
monkeypatch.setattr(conductor.threading.Thread, "start", lambda _: None)
def forbidden(*args, **kwargs):
raise AssertionError("startup must not access desktop settings")
monkeypatch.setattr("builtins.open", forbidden)
monkeypatch.setattr(Path, "open", forbidden)
instance = conductor.Conductor()
instance.start()
assert instance.agent is agent
assert instance.model_snapshot()["effective"] == 1
assert agent.next_llm_calls == [1]