conductor: push unread user msgs into prompt (mark read after turn), identity in sys prompt, expose subagent llm/log

This commit is contained in:
Liang Jiaqing committed 2026-10-30 10:16:02 +08:00
1 parent e4d956f020
commit 5dee7f4bd3
1 file changed
+32 -22
+32 -22
View File
@@ -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: