import threading, sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from fastapi import FastAPI, Header, HTTPException, Query, Depends; from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from agentmain import GenericAgent as GA
PORT, API_KEY = int(sys.argv[1]), sys.argv[2]
app, agent, lock = FastAPI(), GA(), threading.Lock()
outputs, stopped = [], True
threading.Thread(target=agent.run, daemon=True).start()
class Req(BaseModel): prompt: str = ""
agent.verbose = False
def require_key(key: str = Query(None), x_api_key: str = Header(None, alias="X-API-Key")):
if API_KEY not in (key, x_api_key): raise HTTPException(404)
def run_task(prompt):
global stopped
segs = [] # 本任务按 turn 索引的分段输出
with lock: task_start = len(outputs)
def flush():
with lock: outputs[task_start:] = segs
try:
dq = agent.put_task(prompt, source="http")
while "done" not in (item := dq.get(timeout=2200)):
outs = item.get("outputs")
if not outs: continue
idx = max(0, int(item.get("turn", 0) or 0) - 1) # turn 1-based → 槽位 0-based
while len(segs) <= idx: segs.append("")
segs[idx] = str(outs[-1]) # 当前 turn
if len(outs) >= 2 and idx >= 1: segs[idx - 1] = str(outs[-2]) # 前一 turn 落定值
flush()
segs = [str(s) for s in item.get("outputs", [])] # done 时全量替换
flush()
finally: stopped = True
@app.post("/put_task")
def put_task(req: Req, _=Depends(require_key)):
global stopped
with lock:
if not stopped: return {"ok": False, "error": "should abort first"}
stopped = False
threading.Thread(target=run_task, args=(req.prompt,), daemon=True).start()
return {"ok": True}
@app.post("/abort")
def abort(_=Depends(require_key)): agent.abort(); return {"ok": True}
@app.post("/input")
def input_task(req: Req, _=Depends(require_key)):
global stopped
if not stopped: agent.intervene = req.prompt; return {"ok": True, "mode": "intervene"}
with lock:
if not stopped: agent.intervene = req.prompt; return {"ok": True, "mode": "intervene"}
stopped = False
threading.Thread(target=run_task, args=(req.prompt,), daemon=True).start()
return {"ok": True, "mode": "task"}
@app.get("/output")
def get_output(k: int = Query(5), _=Depends(require_key)):
with lock: r = outputs[-k:]
return {"stopped": stopped, "output": "\n".join(r),
"history": "\n".join(str(h) for h in agent.history)}
@app.get("/llm")
def llm_ep(llm_no: int = Query(None), _=Depends(require_key)):
if llm_no is not None:
agent.next_llm(llm_no)
return {"llm_no": agent.llm_no, "name": agent.get_llm_name(),
"llms": [{"no": i, "name": n, "current": a} for i, n, a in agent.list_llms()]}
@app.get("/sysprompt")
def sysprompt_ep(text: str = Query(None), _=Depends(require_key)):
if text is not None:
agent.extra_sys_prompts = [text] if text else []
return {"extra_sys_prompts": agent.extra_sys_prompts}
HELP = """GA HTTP 操作协议（所有请求带 ?key=API_KEY，或 Header X-API-Key）
GET /output?k=N 查看状态：{stopped, output(末N条), history}。stopped=true 表示空闲
POST /input {prompt} 下发指令：空闲时作为新任务，忙时作为中途干预(intervene)
POST /abort 中止当前任务
GET /llm[?llm_no=N] 查/切模型：返回 {llm_no,name,llms:[{no,name,current}]}
GET /sysprompt[?text]查/设附加系统提示(extra_sys_prompts)，text 为空则清空
纠偏流程：先 GET /output 读 history 判断状态→需要时 POST /input 注入纠偏指令"""
@app.get("/help")
def help_ep(_=Depends(require_key)): return {"help": HELP}
@app.get("/")
def ui():
return HTMLResponse(f"""
GA Monitor
● Loading...
""")
if __name__ == "__main__":
import uvicorn; uvicorn.run(app, host="0.0.0.0", port=PORT)