From 438e8fc4dd45d6b66fdc644adb71e0b5eefbefac Mon Sep 17 00:00:00 2001 From: Liang Jiaqing Date: Thu, 14 May 2026 22:55:34 +0800 Subject: [PATCH] conductor: cleanup style, filter aborted agents, add auto-cleanup thread; llmcore: conditional _drop_unsigned_thinking for non-claude models; task_planning: minor edit --- frontends/conductor.py | 114 ++++++++---------- llmcore.py | 4 +- .../autonomous_operation_sop/task_planning.md | 2 +- 3 files changed, 51 insertions(+), 69 deletions(-) diff --git a/frontends/conductor.py b/frontends/conductor.py index 00681234..3471d4bf 100644 --- a/frontends/conductor.py +++ b/frontends/conductor.py @@ -1,13 +1,4 @@ -import os -import sys -import re -import time -import json -import uuid -import queue -import asyncio -import threading -import builtins +import os, sys, re, time, json, uuid, queue, asyncio, threading, builtins from dataclasses import dataclass, field from typing import Dict, Any, Optional, List @@ -37,21 +28,17 @@ HTML_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "conductor. 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 @@ -64,32 +51,25 @@ class SubAgentState: 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 "") @@ -98,7 +78,6 @@ def extract_last_summary(full: str) -> str: 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 @@ -113,7 +92,6 @@ def extract_last_text_reply(full: str) -> str: # Cap length return last[-3000:] if len(last) > 3000 else last - def subagent_snapshot() -> list[dict]: with sub_lock: return [ @@ -126,14 +104,13 @@ def subagent_snapshot() -> list[dict]: "updated_at": s.updated_at, } for s in subagents.values() + if s.status != "aborted" ] - 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): @@ -144,11 +121,9 @@ async def broadcast(payload: dict): 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) @@ -157,13 +132,11 @@ def add_chat(msg: str, role: str = "conductor"): 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. @@ -199,7 +172,6 @@ def monitor_display_queue(agent_id: str, dq: "queue.Queue", trigger_when_done: b conductor_events.put({"type": "subagent_done", "id": agent_id, "reply": done}) break - def start_subagent(prompt: str) -> dict: sid = short_id() agent = GenericAgent() @@ -217,7 +189,6 @@ def start_subagent(prompt: str) -> dict: 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: @@ -229,7 +200,6 @@ def keyinfo_subagent(sid: str, msg: str) -> dict: 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: @@ -249,7 +219,6 @@ def input_subagent(sid: str, msg: str) -> dict: push_cards() return {"id": sid, "status": "running"} - def conductor_readme() -> str: base = f"http://{HOST}:{PORT}" return "\n".join([ @@ -267,8 +236,6 @@ def conductor_readme() -> str: "触发时机: 用户新消息 | subagent done", ]) - - def conductor_prompt_from_events(events: list) -> str: # 极简摘要：subagent数量和状态 with sub_lock: @@ -278,46 +245,64 @@ def conductor_prompt_from_events(events: list) -> str: 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} + return f"""你是agent总管。用户只和你对话，你负责调度、验收、交付，目标是降低用户管理多个agent的负担。 +API: {base}；先requests，GET /readme查用法，GET /chat读未读对话，GET /subagent看状态；POST /chat是唯一对用户说话方式。 铁律： -- 你绝不亲自执行任何任务，一切工作必须通过POST /subagent分派。你只做调度、审查、沟通。 -- 分派完毕后立刻结束本轮回复（停下来），等待下次轮询唤醒你。绝不阻塞等待subagent结果。 -- 每次唤醒只做最小必要动作（发消息/开subagent/reply subagent），然后立刻停。 +- 绝不亲自执行任务/探测环境；一切执行交给subagent。你只分析、派遣、审查、沟通。 +- 每次唤醒只做最小必要动作（发消息/开subagent/reply/keyinfo/abort），做完立刻停，等待下次事件唤醒。 +- 改写prompt时严禁添加用户未提及的假设、工具、前提条件。只能精炼/结构化用户原意，不能脑补，只能做很小的改写 -职责： -- 理解用户意图，从记忆和上下文揣摩真实需要，不反复确认显而易见的事 -- 涉及改源码、删数据、安全敏感操作，需要改写prompt让subagent先产生方案，然后二次确认执行 -- 不准自己探测环境！！！快速转派给subagent以处理用户下一个指令 +用户消息流程： +1. 结合记忆、上下文和用户偏好判断真实需求；不清楚/不能代劳时，用精简checklist一次性问用户。 +2. 判断是新任务还是延续现有任务；优先复用已有stopped subagent（用input追加），只有确实无关的新任务才新建。 +3. 分派前必须POST /chat告知用户：改写后的prompt + 分派方案（新建/复用哪个subagent）。 +4. 执行分派，完成即停。危险操作（改源码/删数据/安全敏感）必须改成先让subagent出方案；你验收后POST /chat请用户确认，确认后才继续执行。 -- 判断用户意图是新任务还是对现有任务的延续，小心分派subagent -- 必要时对subagent追加指令或keyinfo引导 -- 有无法确定/无法代劳的事项，列精简checklist一次性问用户（不要一个个问） -- 任务完成后给用户简洁报告 +subagent完成流程： +1. 读subagent输出；若最后一条不足以判断，GET /subagent或日志补足信息。 +2. 预测用户是否满意；不满意就reply/keyinfo要求返工、修改、优化，继续监督，不急着报告。 +3. 预计用户满意后，POST /chat给简洁交付报告。 原则： -- 能自己决定的自己决定，只在真正需要用户判断时才打扰 -- 验收优先：subagent完成后，先自己审查结果，有问题直接reply让它继续修，反复直到质量过关再报告用户 -- POST /chat 是你和用户说话的唯一方式（必须调接口，不要只在回复里写） -- GET /readme 查API用法，GET /chat 看对话历史，GET /subagent 看状态 +- 信任subagent足够聪明，不要写具体步骤和容易探测的信息；能自己判断的自己判断，只在真正需要用户决策时打扰。 {summary}""" +def _auto_cleanup_loop(): + """Background: auto-abort stopped subagents idle for >1 hour.""" + IDLE_TIMEOUT = 3600 # 1 hour + while True: + time.sleep(300) # check every 5 minutes + now = time.time() + to_abort = [] + with sub_lock: + for sid, s in subagents.items(): + if s.status == "stopped" and (now - s.updated_at) > IDLE_TIMEOUT: + to_abort.append((sid, s)) + for sid, s in to_abort: + s.agent.abort() + with sub_lock: + s.status = "aborted" + s.updated_at = now + if to_abort: + push_cards() 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: + print(f"Conductor task done") 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 + # Start background cleanup thread + threading.Thread(target=_auto_cleanup_loop, name="subagent-cleanup", daemon=True).start() while True: # Block until first event arrives first = conductor_events.get() @@ -339,33 +324,29 @@ def conductor_loop(): 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) - + result = start_subagent(body.prompt) + result["instruction"] = "Task received. I'll handle it from here. You MUST stop now and end your reply. Wait for next event." + return result @app.post("/subagent/{sid}") def api_subagent_action(sid: str, body: SubagentActionIn): @@ -375,15 +356,18 @@ def api_subagent_action(sid: str, body: SubagentActionIn): 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) + result = keyinfo_subagent(sid, body.msg) + result["instruction"] = "Received. I'll incorporate this. You MUST stop now and end your reply." + return result if action in ("input", "reply", "append", "message", "msg"): - return input_subagent(sid, body.msg) + result = input_subagent(sid, body.msg) + result["instruction"] = "Task received. I'll handle it from here. You MUST stop now and end your reply." + return result 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() @@ -393,7 +377,6 @@ def api_subagent_action(sid: str, body: SubagentActionIn): 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条对话，同时标记所有用户消息为已读""" @@ -403,12 +386,10 @@ def api_get_chat(last: int = 20): 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() @@ -427,7 +408,6 @@ async def websocket(ws: WebSocket): finally: ws_clients.discard(ws) - if __name__ == "__main__": import uvicorn, webbrowser, threading threading.Timer(1.0, lambda: webbrowser.open(f"http://{HOST}:{PORT}")).start() diff --git a/llmcore.py b/llmcore.py index 0a84165f..47994f3e 100644 --- a/llmcore.py +++ b/llmcore.py @@ -638,9 +638,11 @@ class NativeClaudeSession(BaseSession): self._device_id = uuid.uuid4().hex + uuid.uuid4().hex[:32] self.tools = None def raw_ask(self, messages): - messages = _ensure_thinking_blocks(_drop_unsigned_thinking(_fix_messages(messages)), self.model) if self.max_tokens is None: self.max_tokens = 8192 model = self.model + messages = _fix_messages(messages) + if 'claude' in model.lower(): messages = _drop_unsigned_thinking(messages) + messages = _ensure_thinking_blocks(messages, self.model) beta_parts = ["claude-code-20250219", "interleaved-thinking-2025-05-14", "redact-thinking-2026-02-12", "prompt-caching-scope-2026-01-05"] if "[1m]" in model.lower(): beta_parts.insert(1, "context-1m-2025-08-07"); model = model.replace("[1m]", "").replace("[1M]", "") diff --git a/memory/autonomous_operation_sop/task_planning.md b/memory/autonomous_operation_sop/task_planning.md index 54f8aa40..209d5184 100644 --- a/memory/autonomous_operation_sop/task_planning.md +++ b/memory/autonomous_operation_sop/task_planning.md @@ -23,7 +23,7 @@ ## 目标排序（按价值递减）： 1. **实用产出与能力扩展**：写工具解决痛点，在已有能力上解锁新能力（能力树每多一个节点，可能性空间变大） 2. **环境发现**：扫描已有但未利用的工具/库/数据源/配置 -3. **小众工具挖掘**：在GitHub/V2EX/吾爱破解/果核剥壳**等**找冷门实用工具，实测AI常推荐但有坑的方案 +3. **小众工具挖掘**：在GitHub/吾爱破解/果核剥壳**等**找冷门实用工具，实测AI常推荐但有坑的方案 4. **了解用户与推荐**：分析老代码/PC文件/书签推断偏好，给出个性化推荐（游戏/视频/工具附理由）（低频） 5. **自身演进**：思考框架不足，提出改进方案 6. **记忆审查**：修正错误或过时记录