github-featured/lsdefine--genericagent
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

conductor: cleanup style, filter aborted agents, add auto-cleanup thread; llmcore: conditional _drop_unsigned_thinking for non-claude models; task_planning: minor edit

Browse Source
This commit is contained in:
Liang Jiaqing
2026-05-14 22:55:34 +08:00
parent b4f0763cc8
commit 438e8fc4dd
3 changed files with 51 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontends/conductor.py
+47 -67
View File
@@ -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'<summary>(.*?)</summary>\s*', re.DOTALL)
def extract_last_summary(full: str) -> str:
"""Extract the latest <summary> 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()
llmcore.py
+3 -1
View File
@@ -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]", "")
memory/autonomous_operation_sop/task_planning.md
+1 -1
View File
@@ -23,7 +23,7 @@
## 目标排序（按价值递减）：
1. **实用产出与能力扩展**：写工具解决痛点，在已有能力上解锁新能力（能力树每多一个节点，可能性空间变大）
2. **环境发现**：扫描已有但未利用的工具/库/数据源/配置
3. **小众工具挖掘**：在GitHub/V2EX/吾爱破解/果核剥壳**等**找冷门实用工具，实测AI常推荐但有坑的方案
3. **小众工具挖掘**：在GitHub/吾爱破解/果核剥壳**等**找冷门实用工具，实测AI常推荐但有坑的方案
4. **了解用户与推荐**：分析老代码/PC文件/书签推断偏好，给出个性化推荐（游戏/视频/工具附理由）（低频）
5. **自身演进**：思考框架不足，提出改进方案
6. **记忆审查**：修正错误或过时记录