hub: switch LLM channel (op llm / GET llms, POST llm) via agent.list_llms/next_llm only

This commit is contained in:
Liang Jiaqing committed 2026-09-11 10:18:38 +08:00
1 parent f6e5657fbe
commit 4913a6162b
2 files changed
+46 -14

No files matched your search

+25 -6
View File
@@ -46,7 +46,8 @@ body{margin:0;height:100vh;display:flex;background:var(--bg);color:var(--fg);
#tip{min-height:16px;margin-top:5px;font-size:12px;color:var(--dim)}
#tip.err{color:#c0392b}
#main>.hd{font-size:18px;margin-bottom:18px;padding-bottom:12px;
border-bottom:1px solid var(--line)}
border-bottom:1px solid var(--line);display:flex;align-items:center;gap:10px}
#main>.hd .ttl{flex:1} #main>.hd .llm{font-size:12px;max-width:180px}
.msg{margin:0 0 14px}
.msg.user{background:var(--user);border-radius:12px;padding:10px 14px;
white-space:pre-wrap;max-width:70%;margin-left:auto}
@@ -116,7 +117,7 @@ async function loadPeers(){
d.className = 'peer' + (p.name === cur ? ' on' : '');
d.dataset.name = p.name;
d.innerHTML = `<div class="t">${esc(p.title || p.name)}</div>
<div class="m">${esc(p.name)} · ${p.n_msgs} 条</div>`;
<div class="m">${esc(p.name)} · ${p.n_msgs} 条${p.llm ? ' · ' + esc(p.llm) : ''}</div>`;
d.onclick = () => open_(p.name);
el.appendChild(d);
}
@@ -156,8 +157,9 @@ async function sync(first){
if (!first && (olds.length > total || (olds.length && t0 && olds[0]._input !== t0.input)))
return open_(name);
const atEnd = m.scrollHeight - m.scrollTop - m.clientHeight < 80;
if (first || !m.querySelector('.hd')) m.innerHTML = '<div class="hd"></div>';
m.querySelector('.hd').textContent = r.title || name;
if (first || !m.querySelector('.hd')) m.innerHTML = '<div class="hd"><span class="ttl"></span><select class="llm" title="切换 LLM"></select></div>';
m.querySelector('.hd .ttl').textContent = r.title || name;
llmSel(m.querySelector('.hd .llm'), name, r.llm);
for (const t of tasks){
const el = m.querySelector(`.task[data-i="${t.i}"]`);
@@ -228,6 +230,23 @@ function segEl(name, i, s){
return det;
}
// LLM channel switch: the <select> lists the host's channels on first click, POST llm {no} on change
function llmSel(sel, name, llm){
if (sel._llm === llm) return; sel._llm = llm;
sel.innerHTML = `<option>${esc(llm || '?')}</option>`; sel.disabled = !llm;
sel.onmousedown = async () => {
if (sel.options.length > 1) return;
const r = await api(`/api/${encodeURIComponent(name)}/llms`);
if (r.error) return;
sel.innerHTML = r.items.map((n, i) => `<option value="${i}"${i === r.cur ? ' selected' : ''}>${esc(n)}</option>`).join('');
};
sel.onchange = async () => {
const r = await post('llm', {no: +sel.value});
say(r.error || '已切换 ' + sel.options[sel.selectedIndex].text, !!r.error);
if (r.error) sel._llm = null; // next sync repaints the true state
};
}
// ---- composer: send a task / abort the running one (both are idempotent-safe) ----
const box = $('#box'), tip = $('#tip');
@@ -235,11 +254,11 @@ function say(t, err){ tip.textContent = t || ''; tip.classList.toggle('err', !!e
function enable(on){ for (const e of [box, $('#send'), $('#stop')]) e.disabled = !on;
box.placeholder = on ? 'Enter 发送 / Shift+Enter 换行' : '先选择一个会话…'; }
async function post(path){ // -> {ok} | {error, code}
async function post(path, body){ // -> {ok} | {error, code}
try {
const r = await fetch(`/api/${encodeURIComponent(cur)}/${path}`, {
method: 'POST', headers: {'Content-Type': 'application/json'},
body: JSON.stringify({text: box.value})});
body: JSON.stringify(body || {text: box.value})});
return await r.json();
} catch(e){ return {error: '网络错误'}; }
}
+21 -8
View File
@@ -1,10 +1,11 @@
"""GA Hub: `import hub` -> client (silent if no server); `python hub.py` -> WS server.
hub.connect(agent, 'stapp') # or override any hook: connect(a, n, put_task=, get_outputs=, abort=)
hub.connect(agent, 'stapp') # or override any hook: connect(a, n, put_task=, get_outputs=, abort=, llm=)
default put -> 'busy' if agent.is_running, else parks plain text in agent._hub_inbox for the UI
HTTP (errors = {'error','code'} + status: offline/gone 404, busy 409, timeout 504, nosupport 501, badop 400):
GET peers -> [{name,title,n_msgs,sig}] | {name}/messages?detail=1&sig= -> {title,tasks:[{i,input,steps:
[{j,title,n}]}],sig} or {same:1,sig} | {name}/seg/{i}/{j}?off=N -> {content,off,n} (step bodies, tailable)
POST {name}/put {"text":..} -> {ok:1} | {name}/abort -> {ok:1}
GET {name}/llms -> {cur,items:[name]} | POST {name}/llm {"no":N} -> {ok:1,cur} (switch LLM channel, busy 409)
"""
import os, re, sys, json, time, asyncio, threading, random, hmac, hashlib
PORT = int(os.environ.get('GA_HUB_PORT', 19736))
@@ -15,8 +16,8 @@ TITLE_MIN = 12 # a too short last input gets the previous one
NOISE = re.compile(r'\**LLM Running \(Turn \d+\) \.\.\.\**|`{3,}.*?`{3,}|<thinking>.*?</thinking>', re.DOTALL)
class HubClient:
def __init__(self, name, put_task=None, get_outputs=None, abort=None, state=None, on_ev=None, sub=(), fixed=False):
self.name, self.put_task, self.get_outputs, self.abort = name, put_task, get_outputs, abort
def __init__(self, name, put_task=None, get_outputs=None, abort=None, state=None, on_ev=None, sub=(), fixed=False, llm=None):
self.name, self.put_task, self.get_outputs, self.abort, self.llm = name, put_task, get_outputs, abort, llm
self.state, self.on_ev, self.sub, self.fixed = state or dict, on_ev, list(sub), fixed # fixed -> stable name
self._tc, self._ws, self._lp = {}, None, None # (i,j) -> (fp, title): a finished step is never re-scanned
def emit(self, topic, data=None, to=None): # thread-safe fire-and-forget from the host thread
@@ -31,7 +32,7 @@ class HubClient:
try:
async with websockets.connect(URL, open_timeout=3, max_size=None) as ws:
self._ws, self._lp = ws, asyncio.get_running_loop()
caps = [k for k, v in (('get', self.get_outputs), ('put', self.put_task), ('abort', self.abort)) if v]
caps = [k for k, v in (('get', self.get_outputs), ('put', self.put_task), ('abort', self.abort), ('llm', self.llm)) if v]
await ws.send(json.dumps({'op': 'hello', 'name': self.name, 'pid': os.getpid(),
'fixed': self.fixed, 'caps': caps, 'sub': self.sub}))
async for raw in ws: await self._on_cmd(ws, json.loads(raw))
@@ -76,6 +77,7 @@ class HubClient:
if op in ('get', 'seg'): data = await asyncio.to_thread(self._build if op == 'get' else self._seg, c)
elif op == 'put_task': data = (c.get('text') and await asyncio.to_thread(self.put_task, c['text'])) or {'ok': 1}
elif op == 'abort': data = (await asyncio.to_thread(self.abort) or {'ok': 1}) if self.abort else {'error': 'no abort hook', 'code': 'nosupport'}
elif op == 'llm': data = await asyncio.to_thread(self.llm, c.get('no')) if self.llm else {'error': 'no llm hook', 'code': 'nosupport'}
await ws.send(json.dumps({'op': 'r', 'id': c.get('id'), 'name': self.name, 'data': data}, default=str))
def serve():
@@ -91,7 +93,7 @@ def serve():
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
creationflags=0x08000008 if os.name == 'nt' else 0) # DETACHED | NO_WINDOW
def connect(agent, name=None, put_task=None, get_outputs=None, abort=None, fold=None):
def connect(agent, name=None, put_task=None, get_outputs=None, abort=None, fold=None, llm=None):
"""One line to wire a GA host: `hub.connect(agent, 'stapp')`; any hook can still be overridden.
Default put refuses while the agent is busy (a remote must not cut in line), else it parks
plain text in agent._hub_inbox; the UI feeds it through its own input entrance when idle,
@@ -101,13 +103,20 @@ def connect(agent, name=None, put_task=None, get_outputs=None, abort=None, fold=
if not isinstance(text, str): return {'error': 'text must be a string', 'code': 'badop'}
if getattr(agent, 'is_running', False): return {'error': f'peer {name} is busy', 'code': 'busy'}
agent._hub_inbox.append(text) # no put_task here: the UI's unified entrance sends it
def _llm(no=None): # no=None -> list; no=N -> switch channel (same as the stapp sidebar)
if no is None: return {'cur': agent.llm_no, 'items': [n for _, n, _ in agent.list_llms()]}
if getattr(agent, 'is_running', False): return {'error': f'peer {name} is busy', 'code': 'busy'}
if not 0 <= no < len(agent.list_llms()): return {'error': f'no llm #{no}', 'code': 'badop'}
agent.next_llm(no); return {'ok': 1, 'cur': no}
try:
try: serve() # best effort: bring up a local hub if none is listening
except Exception: pass
if not hasattr(agent, '_hub_inbox'): agent._hub_inbox = []
agent._hub = HubClient(name or getattr(agent, 'name', 'agent'), put_task or _put,
get_outputs or (lambda: agent.all_outputs), abort or agent.abort,
state=lambda: {'run': bool(getattr(agent, 'is_running', False))})
llm=llm or (_llm if hasattr(agent, 'list_llms') else None),
state=lambda: {'run': bool(getattr(agent, 'is_running', False)),
'llm': getattr(getattr(agent, 'llmclient', None), 'name', '')})
return agent._hub.start()
except Exception: return None
@@ -175,10 +184,10 @@ if __name__ == '__main__':
if 'get' not in caps: rows.append({'name': n, 'title': '', 'n_msgs': 0, 'sig': None, 'caps': caps}); continue
if r is None or r.get('error') or (r.get('same') and n in pcache): # busy/skipped/idle -> last row
rows.append({**(pcache.get(n) or {'name': n, 'title': '?', 'n_msgs': 0, 'sig': None}),
'caps': caps, 'run': (r or {}).get('run')}); continue
'caps': caps, 'run': (r or {}).get('run'), 'llm': (r or {}).get('llm')}); continue
msgs = sum(1 for t in r.get('tasks', []) if not (t.get('input') or '').lstrip().startswith('/'))
rows.append({'name': n, 'title': r.get('title', '?'), 'n_msgs': msgs, 'sig': r.get('sig'),
'caps': caps, 'run': r.get('run')})
'caps': caps, 'run': r.get('run'), 'llm': r.get('llm')})
if r.get('sig'): pcache[n] = rows[-1]
for n in [x for x in pcache if x not in peers]: pcache.pop(n, None)
if psig is None: return rows
@@ -195,6 +204,10 @@ if __name__ == '__main__':
return out(await ask(name, {'op': 'put_task', 'text': t}))
@app.post('/api/{name}/abort')
async def api_abort(name: str): return out(await ask(name, {'op': 'abort'}))
@app.get('/api/{name}/llms')
async def api_llms(name: str): return out(await ask(name, {'op': 'llm'}))
@app.post('/api/{name}/llm')
async def api_llm(name: str, body: dict): return out(await ask(name, {'op': 'llm', 'no': int(body.get('no', -1))}))
# ---- optional P2P pairing; delete this block to remove it completely ----
try:
from hub_p2p import install as _install_p2p