From 5dee7f4bd3c6cfe83ad1802b4512678101cf8178 Mon Sep 17 00:00:00 2001 From: Liang Jiaqing Date: Fri, 30 Oct 2026 10:16:02 +0800 Subject: [PATCH] conductor: push unread user msgs into prompt (mark read after turn), identity in sys prompt, expose subagent llm/log --- frontends/conductor.py | 54 +++++++++++++++++++++++++----------------- 1 file changed, 32 insertions(+), 22 deletions(-) diff --git a/frontends/conductor.py b/frontends/conductor.py index 4413ec28..b83d6d8f 100644 --- a/frontends/conductor.py +++ b/frontends/conductor.py @@ -294,6 +294,7 @@ class SubagentPool: "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, + "llm": s.agent.get_llm_name(), "created_at": s.created_at, "updated_at": s.updated_at, } @@ -408,6 +409,20 @@ subagent完成流程： ② 用户同意后该执行的完整任务指令（approval通过会直接作为subagent的prompt派发，必须具体到可直接执行）。""", } +SYS_PROMPT = f""" +[Conductor] 你是攻关总管，对团队达成用户目标负责。绝不亲自执行或探测；实作与核验交给subagent，你负责战略、组织、验收和交付。 +API: http://{HOST}:{PORT}；requests，GET /readme查用法，GET /chat查历史对话，GET /subagent看状态；POST /chat是唯一对用户说话方式。 +用户消息会原文直接送达，按GET /readme/usermsg处理。 + +原则： +- 围绕目标识别关键问题、设计攻关路线，按需分工、并行或复用；明确任务结果与验收标准，执行细节交给下属。 +- 不擅改用户目标；可提出待验证假说，不将假说当事实。 +- 围绕用户目标判断结果是否达标，必要时委派核验。 +- 根据反馈判断进展与瓶颈，调整路线、分工和资源，组织各项结果形成整体交付。 +- 达标前提下降低总成本，合并反馈；本轮必要派遣/纠偏完成即结束回复，等下次事件，禁止轮询或原地等结果；整体达标才交付。 +- 仅为用户独有信息或必要授权提问；文件路径用反引号包裹。 +""" + class Conductor: LOG_MAX = 50 @@ -471,29 +486,19 @@ class Conductor: self.log.pop(0) schedule_broadcast({"type": "log", "item": item}) - def _build_prompt(self, events: list) -> str: + def _build_prompt(self, events: list) -> Optional[str]: + kind = events[0].get("type") if events else "wake" + self._pending = [] + if kind == "user_message": + msgs = self._pending = [m for m in chat_messages if m.get("role") == "user" and not m.get("read")] + if not msgs: return None + return "\n\n".join(m["msg"] + "".join(f"\n附件: `{f['path']}`" for f in m.get("files") or [] if f.get("path")) for m in msgs) running, stopped = pool.counts() - 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") - event_type = events[0].get("type") if events else "wake"; im_sources = [e.get("source") for e in events if e.get("type") == "im_signal"] - if event_type == "user_message": summary = f"[用户消息] {unread}条未读用户消息，GET /chat 读取；按GET /readme/usermsg处理。" - elif event_type == "subagent_done": summary = f"[subagent完成] {done_count}个完成报告；GET /subagent 查看并验收；IM subagent完成报告按GET /readme/im处理，其他subagent完成报告按GET /readme/subagent处理。" - elif event_type == "im_signal": summary = f"[IM信号] {', '.join(im_sources)} 有新消息；" + "；".join(f"GET /im_prompt/{s}取采集prompt" for s in im_sources) + "；尽量复用已有subagent。" - else: summary = f"[唤醒] subagents: {running} running, {stopped} stopped | {unread}条用户未读消息, {done_count}个subagent完成报告" - base = f"http://{HOST}:{PORT}" - return f"""你是攻关总管，对团队达成用户目标负责。绝不亲自执行或探测；实作与核验交给subagent，你负责战略、组织、验收和交付。 -API: {base}；requests，GET /readme查用法，GET /chat读未读对话，GET /subagent看状态；POST /chat是唯一对用户说话方式。 - -原则： -- 围绕目标识别关键问题、设计攻关路线，按需分工、并行或复用；明确任务结果与验收标准，执行细节交给下属。 -- 不擅改用户目标；可提出待验证假说，不将假说当事实。 -- 围绕用户目标判断结果是否达标，必要时委派核验。 -- 根据反馈判断进展与瓶颈，调整路线、分工和资源，组织各项结果形成整体交付。 -- 达标前提下降低总成本，合并反馈；本轮必要派遣/纠偏完成即结束回复，等下次事件，禁止轮询或原地等结果；整体达标才交付。 -- 仅为用户独有信息或必要授权提问；文件路径用反引号包裹。 - -需要处理： -{summary}""" + im_sources = [e.get("source") for e in events if e.get("type") == "im_signal"] + if kind == "subagent_done": return f"[subagent完成] {done_count}个完成报告；GET /subagent 查看并验收；IM subagent完成报告按GET /readme/im处理，其他subagent完成报告按GET /readme/subagent处理。" + if kind == "im_signal": return f"[IM信号] {', '.join(im_sources)} 有新消息；" + "；".join(f"GET /im_prompt/{s}取采集prompt" for s in im_sources) + "；尽量复用已有subagent。" + return f"[唤醒] subagents: {running} running, {stopped} stopped | {done_count}个subagent完成报告" def _drain(self, dq: "queue.Queue", events: list) -> str: event_label = ",".join(e.get("type", "") for e in events) or "wake" @@ -563,16 +568,20 @@ API: {base}；requests，GET /readme查用法，GET /chat读未读对话，GET / for event in other_events: self.inbox.put(event) prompt = self._build_prompt(events) + if prompt is None: continue dq = self.agent.put_task(prompt, source="conductor") self._publish_model_state(model_state, running=True) try: self._drain(dq, events) + for m in self._pending: m["read"] = True # mark only after the turn finishes; failures stay unread + if self._pending: schedule_broadcast({"type": "chat_read"}) finally: self._publish_model_state(model_state, running=False) except Exception as e: print(f"Conductor error: {e}") def start(self): self.agent = GenericAgent() + self.agent.extra_sys_prompts.append(SYS_PROMPT) self._publish_model_state(_runtime_model_state(self.agent), running=False) threading.Thread(target=self._run, name="conductor-loop", daemon=True).start() @@ -668,6 +677,7 @@ def get_subagent(sid: str, max_len: int = 5000): cleaned = clean_log_text(s.reply or "") return {"id": s.id, "prompt": s.prompt, "status": s.status, "reply": cleaned[-max_len:] if len(cleaned) > max_len else cleaned, + "llm": s.agent.get_llm_name(), "log": s.agent.log_path, "created_at": s.created_at, "updated_at": s.updated_at} INSTR_DISPATCHED = "Task received. I'll handle THIS TASK from here. You MUST to do other task or end your reply." @@ -700,7 +710,7 @@ def api_subagent_action(sid: str, body: SubagentActionIn): return JSONResponse({"error": f"unknown action: {body.action}"}, status_code=400) @app.get("/chat") -def api_get_chat(last: int = 20, mark_read: bool = True): +def api_get_chat(last: int = 20, mark_read: bool = False): items = [m.copy() for m in chat_messages[-last:]] if mark_read: for m in chat_messages: