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

hub: persist token to temp/.hub_token; add intervene/keyinfo inject op + hub.call/peers helper

Browse Source
This commit is contained in:
Liang Jiaqing committed 2026-09-29 09:28:08 +08:00
1 parent e86ca723a9
commit 9dcbf5adf0
1 file changed
+43 -5
Download Patch File Download Diff File Expand all files Collapse all files
frontends/hub.py
+43 -5
View File
@@ -16,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, llm=None):
self.name, self.put_task, self.get_outputs, self.abort, self.llm = name, put_task, get_outputs, abort, llm
def __init__(self, name, put_task=None, get_outputs=None, abort=None, state=None, on_ev=None, sub=(), fixed=False, llm=None, inject=None):
self.name, self.put_task, self.get_outputs, self.abort, self.llm, self.inject = name, put_task, get_outputs, abort, llm, inject
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
@@ -32,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), ('llm', self.llm)) if v]
caps = [k for k, v in (('get', self.get_outputs), ('put', self.put_task), ('abort', self.abort), ('llm', self.llm), ('inject', self.inject)) 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))
@@ -78,8 +78,27 @@ class HubClient:
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'}
elif op == 'inject': data = (await asyncio.to_thread(self.inject, c.get('kind'), c.get('text')) or {'ok': 1}) if self.inject else {'error': 'no inject hook', 'code': 'nosupport'}
await ws.send(json.dumps({'op': 'r', 'id': c.get('id'), 'name': self.name, 'data': data}, default=str))
_ROUTES = {'put': ('POST', 'put'), 'abort': ('POST', 'abort'), 'intervene': ('POST', 'intervene'), 'keyinfo': ('POST', 'keyinfo'),
'llm': ('POST', 'llm'), 'llms': ('GET', 'llms'), 'messages': ('GET', 'messages')}
def call(name=None, op='peers', timeout=20, **kw):
"""Drive any registered GA over HTTP (token from temp/.hub_token), e.g. from another GA:
hub.call('stapp-123', 'intervene', text=...) | 'keyinfo' (running only) | 'put' (idle only) | 'abort' | 'messages' | 'llm', no=N"""
import urllib.request, urllib.parse
tok = os.environ.get('GA_HUB_TOKEN') or open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
'temp', '.hub_token'), encoding='utf-8').read().strip()
if op == 'peers': method, path = 'GET', '/api/peers'
else: method, sub = _ROUTES[op]; path = f'/api/{urllib.parse.quote(name)}/{sub}'
q = {'t': tok, **(kw if method == 'GET' else {})}
req = urllib.request.Request(f'http://127.0.0.1:{WEB_PORT}{path}?{urllib.parse.urlencode(q)}', method=method,
data=json.dumps(kw).encode() if method == 'POST' else None, headers={'Content-Type': 'application/json'})
try: r = urllib.request.build_opener(urllib.request.ProxyHandler({})).open(req, timeout=timeout).read()
except urllib.error.HTTPError as e: r = e.read()
return json.loads(r or b'{}')
def peers(): return call()
def serve():
"""Bring the hub up on demand: any host may spawn it, the port is the lock (a loser just exits).
Detached + windowless, so it outlives its spawner and no console ever flashes."""
@@ -93,7 +112,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, llm=None):
def connect(agent, name=None, put_task=None, get_outputs=None, abort=None, fold=None, llm=None, inject=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,
@@ -108,6 +127,12 @@ def connect(agent, name=None, put_task=None, get_outputs=None, abort=None, fold=
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}
def _inject(kind, text): # mid-run steering, read by ga turn_end as [MASTER] (same as _intervene/_keyinfo files)
attr = {'intervene': 'intervene', 'keyinfo': 'extrakeyinfo'}.get(kind)
if not attr or not isinstance(text, str) or not text: return {'error': 'kind must be intervene|keyinfo, text non-empty', 'code': 'badop'}
if not getattr(agent, 'is_running', False): return {'error': f'peer {name} is idle, use put', 'code': 'idle'}
old = getattr(agent, attr, None)
setattr(agent, attr, f'{old}\n\n{text}' if old else text) # append: several injects in one turn all land
try:
try: serve() # best effort: bring up a local hub if none is listening
except Exception: pass
@@ -115,6 +140,7 @@ def connect(agent, name=None, put_task=None, get_outputs=None, abort=None, fold=
agent._hub = HubClient(name or getattr(agent, 'name', 'agent'), put_task or _put,
get_outputs or (lambda: agent.all_outputs), abort or agent.abort,
llm=llm or (_llm if hasattr(agent, 'list_llms') else None),
inject=inject or (_inject if hasattr(agent, 'intervene') else None),
state=lambda: {'run': bool(getattr(agent, 'is_running', False)),
'llm': getattr(getattr(agent, 'llmclient', None), 'name', '')})
return agent._hub.start()
@@ -125,7 +151,15 @@ if __name__ == '__main__':
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request, Response
from fastapi.responses import FileResponse, JSONResponse; import uvicorn
HERE, OK_ORIGIN = os.path.dirname(os.path.abspath(__file__)), ('', f'http://127.0.0.1:{PORT}', f'http://localhost:{PORT}')
STATUS = {'offline': 404, 'gone': 404, 'timeout': 504, 'nosupport': 501, 'badop': 400, 'busy': 409}
TOKEN_FILE = os.path.join(os.path.dirname(HERE), 'temp', '.hub_token') # detached spawn drops stdout: token lives here
if not os.environ.get('GA_HUB_TOKEN'): # env > persisted file > fresh random
try: TOKEN = open(TOKEN_FILE, encoding='utf-8').read().strip() or TOKEN
except OSError: pass
try:
os.makedirs(os.path.dirname(TOKEN_FILE), exist_ok=True)
with open(TOKEN_FILE, 'w', encoding='utf-8') as f: f.write(TOKEN)
except OSError: pass
STATUS = {'offline': 404, 'gone': 404, 'timeout': 504, 'nosupport': 501, 'badop': 400, 'busy': 409, 'idle': 409}
app, peers, meta, waits, seq, pcache, pdead = FastAPI(), {}, {}, {}, [0], {}, {} # ws / hello / futures / row / probe
@app.middleware('http')
async def guard(req: Request, call_next):
@@ -202,6 +236,10 @@ if __name__ == '__main__':
t = body.get('text', '')
if not isinstance(t, str): return JSONResponse({'error': 'text must be a string', 'code': 'badop'}, 400)
return out(await ask(name, {'op': 'put_task', 'text': t}))
@app.post('/api/{name}/intervene')
async def api_intervene(name: str, body: dict): return out(await ask(name, {'op': 'inject', 'kind': 'intervene', 'text': body.get('text')}))
@app.post('/api/{name}/keyinfo')
async def api_keyinfo(name: str, body: dict): return out(await ask(name, {'op': 'inject', 'kind': 'keyinfo', 'text': body.get('text')}))
@app.post('/api/{name}/abort')
async def api_abort(name: str): return out(await ask(name, {'op': 'abort'}))
@app.get('/api/{name}/llms')