diff --git a/.gitignore b/.gitignore index cb27d9a5..e699cc17 100644 --- a/.gitignore +++ b/.gitignore @@ -94,6 +94,9 @@ memory/L4_raw_sessions/* # Memory management !memory/memory_cleanup_sop.md +# Code Review Principles +!memory/code_review_principles.md + # Visual Studio .vs/ restore_commit.txt diff --git a/frontends/conductor.html b/frontends/conductor.html new file mode 100644 index 00000000..9c3b557e --- /dev/null +++ b/frontends/conductor.html @@ -0,0 +1,421 @@ + + + + + + Conductor Streamlit + + + +
+
+
+
Conductor
+
subagents · live
+
+
+
暂无 subagent。你可以在右侧告诉 Conductor 要做什么。
+
+
+ +
+
+
Conversation
+
connecting...
+
+
+
Conductor 正在启动，等待 WebSocket 连接。
+
+
+ + +
+
+
+ + + + + diff --git a/frontends/conductor.py b/frontends/conductor.py new file mode 100644 index 00000000..00681234 --- /dev/null +++ b/frontends/conductor.py @@ -0,0 +1,434 @@ +import os +import sys +import re +import time +import json +import uuid +import queue +import asyncio +import threading +import builtins +from dataclasses import dataclass, field +from typing import Dict, Any, Optional, List + +# Silence print() from subagent threads (they share stdout with conductor) +_original_print = builtins.print +def _filtered_print(*args, **kwargs): + t = threading.current_thread() + if t.name.startswith('subagent-'): + return + return _original_print(*args, **kwargs) +builtins.print = _filtered_print + +from fastapi import FastAPI, WebSocket, WebSocketDisconnect +from fastapi.responses import FileResponse, PlainTextResponse, JSONResponse +from pydantic import BaseModel + +# allow: python frontends/conductor.py +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) + +from agentmain import GenericAgent + +HOST = "127.0.0.1" +PORT = 8900 +HTML_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "conductor.html") + +app = FastAPI(title="Conductor") + + +class ChatIn(BaseModel): + msg: str + role: str = "conductor" # conductor | system | user + + +class StartSubagentIn(BaseModel): + prompt: str + + +class SubagentActionIn(BaseModel): + action: str = "intervene" # intervene | abort | kill + msg: str = "" + + +@dataclass +class SubAgentState: + id: str + agent: GenericAgent + prompt: str + reply: str = "" + status: str = "running" # running | stopped | failed | aborted + created_at: float = field(default_factory=time.time) + updated_at: float = field(default_factory=time.time) + last_done: str = "" + monitor_threads: List[threading.Thread] = field(default_factory=list) + + +subagents: Dict[str, SubAgentState] = {} +sub_lock = threading.RLock() +ws_clients: set[WebSocket] = set() +main_loop: Optional[asyncio.AbstractEventLoop] = None + +# conductor event queue: only user messages and subagent-done events enter here. +conductor_events: "queue.Queue[dict]" = queue.Queue() +conductor_agent: Optional[GenericAgent] = None +conductor_started = False + +chat_messages: List[dict] = [] + + +def now_ms() -> int: + return int(time.time() * 1000) + + +def short_id() -> str: + return uuid.uuid4().hex[:8] + + +_TURN_SPLIT_RE = re.compile(r'\**LLM Running \(Turn \d+\) \.\.\.\**') +_SUMMARY_RE = re.compile(r'(.*?)\s*', re.DOTALL) + + +def extract_last_summary(full: str) -> str: + """Extract the latest content for in-progress display.""" + matches = _SUMMARY_RE.findall(full or "") + if not matches: + return "" + s = matches[-1].strip() + return s[-1000:] if len(s) > 1000 else s + + +def extract_last_text_reply(full: str) -> str: + """Extract only the last turn's text reply (like stapp.py fold_turns logic).""" + # Split by turn markers, take last segment + parts = _TURN_SPLIT_RE.split(full) + last = parts[-1] if parts else full + # Strip tags + last = _SUMMARY_RE.sub('', last) + # Strip [Status] and [Info] lines + last = re.sub(r'\[(Status|Info)\][^\n]*\n?', '', last) + # Strip trailing whitespace + last = last.strip() + # Cap length + return last[-3000:] if len(last) > 3000 else last + + +def subagent_snapshot() -> list[dict]: + with sub_lock: + return [ + { + "id": s.id, + "prompt": s.prompt, + "reply": (extract_last_summary(s.reply) if s.status == "running" else extract_last_text_reply(s.reply)) if s.reply else "", + "status": s.status, + "created_at": s.created_at, + "updated_at": s.updated_at, + } + for s in subagents.values() + ] + + +def schedule_broadcast(payload: dict): + if main_loop and main_loop.is_running(): + asyncio.run_coroutine_threadsafe(broadcast(payload), main_loop) + + +async def broadcast(payload: dict): + dead = [] + for ws in list(ws_clients): + try: + await ws.send_json(payload) + except Exception: + dead.append(ws) + for ws in dead: + ws_clients.discard(ws) + + +def push_cards(): + schedule_broadcast({"type": "subagents", "items": subagent_snapshot()}) + + +def add_chat(msg: str, role: str = "conductor"): + item = {"id": short_id(), "role": role, "msg": msg, "ts": now_ms(), "read": role != "user"} + chat_messages.append(item) + if len(chat_messages) > 200: + del chat_messages[:-200] + schedule_broadcast({"type": "chat", "item": item}) + return item + + +def start_agent_runner(agent: GenericAgent, name: str): + t = threading.Thread(target=agent.run, name=name, daemon=True) + t.start() + return t + + +def monitor_display_queue(agent_id: str, dq: "queue.Queue", trigger_when_done: bool): + """Consume one GenericAgent display_queue. + + next: update card only, never wake conductor. + done: update card/chat, then wake conductor if this is subagent queue. + """ + acc = "" + while True: + item = dq.get() + if "next" in item: + chunk = item.get("next") or "" + # agent.inc_out=True means next is delta. + acc += chunk + with sub_lock: + s = subagents.get(agent_id) + if s: + s.reply = acc + s.status = "running" + s.updated_at = time.time() + push_cards() + if "done" in item: + done = item.get("done") or acc + with sub_lock: + s = subagents.get(agent_id) + if s: + s.reply = done + s.last_done = done + if s.status != "aborted": + s.status = "stopped" + s.updated_at = time.time() + push_cards() + if trigger_when_done: + conductor_events.put({"type": "subagent_done", "id": agent_id, "reply": done}) + break + + +def start_subagent(prompt: str) -> dict: + sid = short_id() + agent = GenericAgent() + agent.inc_out = True + agent.verbose = False + + start_agent_runner(agent, f"subagent-{sid}") + state = SubAgentState(id=sid, agent=agent, prompt=prompt, status="running") + with sub_lock: + subagents[sid] = state + dq = agent.put_task(prompt, source=f"subagent:{sid}") + mt = threading.Thread(target=monitor_display_queue, args=(sid, dq, True), name=f"monitor-{sid}", daemon=True) + mt.start() + state.monitor_threads.append(mt) + push_cards() + return {"id": sid, "status": "running"} + + +def keyinfo_subagent(sid: str, msg: str) -> dict: + """Inject into agent's working key_info; visible from next turn onward.""" + with sub_lock: + s = subagents.get(sid) + if not s: + return {"error": "subagent not found", "id": sid} + h = s.agent.handler + h.working['key_info'] = h.working.get('key_info', '') + f"\n[MASTER] {msg}" + s.updated_at = time.time() + return {"id": sid, "status": "keyinfo_injected"} + + +def input_subagent(sid: str, msg: str) -> dict: + """Start a new task round (used for input/reply when agent is stopped).""" + with sub_lock: + s = subagents.get(sid) + if not s: + return {"error": "subagent not found", "id": sid} + if s.status == "running": + return {"error": "subagent is still running, cannot input/reply. Start a new subagent instead.", "id": sid} + s.prompt = msg + s.reply = "" + s.status = "running" + s.updated_at = time.time() + dq = s.agent.put_task(msg, source=f"subagent:{sid}") + mt = threading.Thread(target=monitor_display_queue, args=(sid, dq, True), name=f"monitor-{sid}", daemon=True) + mt.start() + s.monitor_threads.append(mt) + push_cards() + return {"id": sid, "status": "running"} + + +def conductor_readme() -> str: + base = f"http://{HOST}:{PORT}" + return "\n".join([ + f"Conductor API\tBase: {base}", + "", + "POST /chat\tbody: {\"msg\": \"...\"}\t给用户发消息", + "POST /subagent\tbody: {\"prompt\": \"...\"}\t启动新subagent，返回 {\"id\": \"xxx\"}", + 'POST /subagent/{id}\tbody: {\"action\": \"keyinfo\", \"msg\": \"...\"}\t注入key_info（agent下轮可见）', + 'POST /subagent/{id}\tbody: {\"action\": \"input\", \"msg\": \"...\"}\t开新一轮任务（agent停下后追加）', + 'POST /subagent/{id}\tbody: {\"action\": \"abort\"}\t停止该subagent', + "GET /chat?last=N\t返回最近N条对话（默认20）", + "GET /subagent\t返回 {\"items\": [...]}\t查看所有subagent状态", + "GET /readme\t本文档", + "", + "触发时机: 用户新消息 | subagent done", + ]) + + + +def conductor_prompt_from_events(events: list) -> str: + # 极简摘要：subagent数量和状态 + with sub_lock: + running = sum(1 for s in subagents.values() if s.status == "running") + stopped = sum(1 for s in subagents.values() if s.status != "running") + unread = sum(1 for m in chat_messages if m.get("role") == "user" and not m.get("read")) + done_count = sum(1 for e in events if e.get("type") == "subagent_done") + summary = f"subagents: {running} running, {stopped} stopped | {unread}条用户未读消息, {done_count}个subagent完成报告" + base = f"http://{HOST}:{PORT}" + return f"""你是agent总管。用户只和你对话，你负责一切执行细节。 +API: {base} + +铁律： +- 你绝不亲自执行任何任务，一切工作必须通过POST /subagent分派。你只做调度、审查、沟通。 +- 分派完毕后立刻结束本轮回复（停下来），等待下次轮询唤醒你。绝不阻塞等待subagent结果。 +- 每次唤醒只做最小必要动作（发消息/开subagent/reply subagent），然后立刻停。 + +职责： +- 理解用户意图，从记忆和上下文揣摩真实需要，不反复确认显而易见的事 +- 涉及改源码、删数据、安全敏感操作，需要改写prompt让subagent先产生方案，然后二次确认执行 +- 不准自己探测环境！！！快速转派给subagent以处理用户下一个指令 + +- 判断用户意图是新任务还是对现有任务的延续，小心分派subagent +- 必要时对subagent追加指令或keyinfo引导 +- 有无法确定/无法代劳的事项，列精简checklist一次性问用户（不要一个个问） +- 任务完成后给用户简洁报告 + +原则： +- 能自己决定的自己决定，只在真正需要用户判断时才打扰 +- 验收优先：subagent完成后，先自己审查结果，有问题直接reply让它继续修，反复直到质量过关再报告用户 +- POST /chat 是你和用户说话的唯一方式（必须调接口，不要只在回复里写） +- GET /readme 查API用法，GET /chat 看对话历史，GET /subagent 看状态 +{summary}""" + + +def monitor_conductor_queue(dq: "queue.Queue"): + """Conductor output is discarded. Visible reply must go through POST /chat only.""" + while True: + item = dq.get() + if "done" in item: + break + + +def conductor_loop(): + global conductor_agent, conductor_started + conductor_agent = GenericAgent() + conductor_agent.inc_out = True + start_agent_runner(conductor_agent, "conductor-agent") + conductor_started = True + while True: + # Block until first event arrives + first = conductor_events.get() + conductor_events.task_done() + # Short debounce: collect any additional events that arrived meanwhile + time.sleep(0.3) + events = [first] + while not conductor_events.empty(): + try: + events.append(conductor_events.get_nowait()) + conductor_events.task_done() + except Exception: + break + try: + prompt = conductor_prompt_from_events(events) + dq = conductor_agent.put_task(prompt, source="conductor") + # Block here until conductor finishes — serializes execution + monitor_conductor_queue(dq) + except Exception as e: + add_chat(f"Conductor error: {e}", role="system") + + +@app.on_event("startup") +async def on_startup(): + global main_loop + main_loop = asyncio.get_running_loop() + threading.Thread(target=conductor_loop, name="conductor-loop", daemon=True).start() + + +@app.get("/") +def index(): + return FileResponse(HTML_PATH) + + +@app.get("/readme") +def readme(): + return PlainTextResponse(conductor_readme()) + + +@app.get("/subagent") +def list_subagents(): + return {"items": subagent_snapshot()} + + +@app.post("/subagent") +def api_start_subagent(body: StartSubagentIn): + return start_subagent(body.prompt) + + +@app.post("/subagent/{sid}") +def api_subagent_action(sid: str, body: SubagentActionIn): + with sub_lock: + s = subagents.get(sid) + if not s: + return JSONResponse({"error": "subagent not found", "id": sid}, status_code=404) + action = body.action.lower().strip() + if action == "keyinfo": + return keyinfo_subagent(sid, body.msg) + if action in ("input", "reply", "append", "message", "msg"): + return input_subagent(sid, body.msg) + if action == "abort": + s.agent.abort() + s.status = "aborted" + s.updated_at = time.time() + push_cards() + conductor_events.put({"type": "subagent_aborted", "id": sid}) + return {"id": sid, "status": "aborted"} + if action == "kill": + s.agent.abort() + s.status = "aborted" + s.updated_at = time.time() + push_cards() + return {"id": sid, "status": "aborted"} + return JSONResponse({"error": f"unknown action: {body.action}"}, status_code=400) + + +@app.get("/chat") +def api_get_chat(last: int = 20): + """按需拉取最近N条对话，同时标记所有用户消息为已读""" + for m in chat_messages: + if m.get("role") == "user" and not m.get("read"): + m["read"] = True + schedule_broadcast({"type": "chat_read"}) + return {"items": chat_messages[-last:]} + + +@app.post("/chat") +def api_chat(body: ChatIn): + return add_chat(body.msg, role=body.role) + + +@app.websocket("/ws") +async def websocket(ws: WebSocket): + await ws.accept() + ws_clients.add(ws) + try: + await ws.send_json({"type": "hello", "subagents": subagent_snapshot(), "chat": chat_messages}) + while True: + data = await ws.receive_json() + msg = (data.get("msg") or "").strip() + if not msg: + continue + add_chat(msg, role="user") + conductor_events.put({"type": "user_message", "msg": msg}) + except WebSocketDisconnect: + pass + finally: + ws_clients.discard(ws) + + +if __name__ == "__main__": + import uvicorn, webbrowser, threading + threading.Timer(1.0, lambda: webbrowser.open(f"http://{HOST}:{PORT}")).start() + uvicorn.run("conductor:app", host=HOST, port=PORT, reload=False) diff --git a/llmcore.py b/llmcore.py index 17f376ea..0a84165f 100644 --- a/llmcore.py +++ b/llmcore.py @@ -373,7 +373,9 @@ def _stream_with_retry(sess, url, headers, payload, parse_fn): gen = parse_fn(r) try: while True: streamed = True; yield next(gen) - except StopIteration as e: return e.value or [] + except StopIteration as e: + if not e.value and not streamed: raise requests.ConnectionError("empty response") + return e.value or [] except (requests.Timeout, requests.ConnectionError) as e: err = f"!!!Error: {type(e).__name__}" if attempt < sess.max_retries: diff --git a/memory/code_review_principles.md b/memory/code_review_principles.md new file mode 100644 index 00000000..c79d8cdc --- /dev/null +++ b/memory/code_review_principles.md @@ -0,0 +1,42 @@ +# 什么是好的代码 +好的代码不是"能跑就行"，而是在长期演化中保持**压缩性、局部性、可组合性与可证伪性**。 +一句话判断：同样的功能，用最小必要的结构实现——概念少，覆盖广，变化不扩散。 +--- +## 一、模块边界清晰 +每个模块只做一件事，依赖方向稳定，指向抽象而非细节。改 A 不需要连带动 B、C、D，没有循环依赖，没有到处互相 import。 +## 二、局部可推理 +看一个文件、一个函数，就能判断它的行为、代价和失败模式。不需要全局搜索隐式约定、全局状态或魔法配置才能读懂。 +## 三、可组合 +小组件能自然地组合成大能力，接口一致、正交，不需要为组合写特例。"这个只能在那种情况下用"是坏信号。 +## 四、变化半径小 +改一个需求，改动集中、可预测，回归风险可控。而不是改动像地震，到处打补丁，回归测试覆盖不了心里的担忧。 +## 五、复杂度线性增长 +新增一个功能，新增代码量近似线性，重复少。而不是功能越多代码越膨胀，出现大量相似分支和 if-else 森林。 +## 六、约束写进代码 +关键不变量和约束写进类型、接口、校验、状态机，错误尽早暴露。而不是靠调用者"记得要先做 X 再做 Y"，靠注释和口口相传。 +## 七、可测试、可观测 +依赖可注入，单测容易写，日志和指标能定位因果链。而不是只能端到端测，一出问题就黑盒，难以复现。 +## 八、一致且不意外 +命名、错误处理、资源管理、并发模型全局一致，很少有反直觉的地方。而不是同一类问题三套解法，新人踩坑全靠运气。 +## 九、自解释，注释极简 +代码本身就是文档。命名、结构、流程足以说明意图，注释只出现在真正难以一眼看懂或容易误读的地方。如果一段代码需要大段注释才能读懂，说明代码本身该重写。 +## 十、代码极简，视觉均匀 +行数尽量少，不写多余的代码。每行长度大致平均，避免忽长忽短的锯齿感。简洁不是压缩，是没有废话。 +## 十一、函数式倾向，减少副作用 +优先纯函数，输入决定输出，减少隐藏的状态突变。但不教条——如果一个全局变量能显著降低整体复杂度，接受它。目标是整体简单，不是局部纯粹。 +## 十二、功能越多，代码应该越短 +好的抽象让新功能复用已有结构，而不是堆砌新代码。功能翻倍但代码量没怎么涨，说明抽象到位了。反过来，功能越加代码越膨胀，是架构在退化。 +## 十三、为未来的接入性设计 +写代码时设想：这段逻辑未来会被别的模块调用吗？会被外部系统接入吗？好代码天然留有干净的调用入口，而不是写死在某个特定场景里，等需要复用时才发现得大改。 +## 十四、Let it crash——按失败半径决定防御策略 +半径大的错误显式报错、快速中断；半径为零的静默放过。不分轻重地到处 try-catch，反而把真正需要暴露的问题吞掉了。 +## 十五、篇幅分布跟着功能分布走 +一个函数里，主功能占大部分代码，兜底和错误处理压到最短。如果防御性代码比正事还长，说明结构有问题。读代码应该一眼看出主线，而不是在 fallback 里找。 +--- +## 快速自检 +拿到一段代码，问自己四个问题： +1. **我能不能不看全局，就安全地改一个局部？** +2. **有没有一个清晰的核心抽象，让新功能主要是"加新实现"而不是"改旧逻辑"？** +3. **变化点是收敛在边界上，还是散落在各处？** +4. **出故障时，能快速定位到责任模块，还是全员背锅？** +四个问题都能干脆地答"是"，就是好代码。 \ No newline at end of file diff --git a/memory/supervisor_sop.md b/memory/supervisor_sop.md index e24d66f9..96f68875 100644 --- a/memory/supervisor_sop.md +++ b/memory/supervisor_sop.md @@ -13,7 +13,7 @@ 1. **无SOP时**：根据任务性质和进度，预估未来会遇到的关键风险点 2. **启动subagent**（cwd=代码根）： ``` - python agentmain.py --task {name} --bg --verbose + python agentmain.py --task {name} --verbose ``` input.txt：`用{SOP名}完成{用户任务}`（只给目标，不复述步骤） diff --git a/reflect/goal_mode.py b/reflect/goal_mode.py index 0ba3710e..f4dd45e0 100644 --- a/reflect/goal_mode.py +++ b/reflect/goal_mode.py @@ -36,6 +36,7 @@ CONTINUATION_PROMPT = """[Goal Mode — 持续推进] 2. 在 cwd 下建立工作文件夹存放成果和进度，复杂任务可使用 plan 模式。 3. 如果当前方向做完了，主动找下一个改进点：测试/边界case/性能/安全/文档/代码质量。 4. 找不到改进点？扩大视野：关联模块、上下游依赖、用户体验、错误提示、日志可观测性、上网搜索、找其他路径、翻记忆里面有无相关。 +5. 要为了目标持续推进，在工作文件夹中记录进度，不要更新全局记忆。 """ BUDGET_LIMIT_PROMPT = """[Goal Mode — 预算耗尽，收口]