Improve session controls and compact LLM logs

This commit is contained in:
Liang Jiaqing committed 2026-07-10 11:40:22 +08:00
1 parent a195bdc101
commit 8ea8807ced
3 files changed
+44 -17

No files matched your search

+26 -8
View File
@@ -42,6 +42,20 @@ def _apply_desktop_model(agent: "GenericAgent") -> None:
print(f"[conductor] failed to apply desktop model #{no}: {e}", file=sys.stderr) print(f"[conductor] failed to apply desktop model #{no}: {e}", file=sys.stderr)
def _select_llm(agent: "GenericAgent", llm: Any) -> bool:
if llm is None or str(llm).strip() == "": return False
q = str(llm).strip()
if isinstance(llm, int) or q.isdigit(): agent.next_llm(int(q)); return True
q = q.lower(); agent.load_llm_sessions()
for i, c in enumerate(agent.llmclients):
vals = []
for fn in (lambda: agent.get_llm_name(c), lambda: agent.get_llm_name(c, model=True), lambda: c.backend.name, lambda: c.backend.model):
try: vals.append(str(fn()).lower())
except Exception: pass
if any(q in v for v in vals): agent.next_llm(i); return True
raise ValueError(f"llm not found: {llm}")
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI): async def lifespan(app: FastAPI):
# 服务启动(事件循环已就绪):捕获 loop 供工作线程跨线程推 WS 广播,并起主agent # 服务启动(事件循环已就绪):捕获 loop 供工作线程跨线程推 WS 广播,并起主agent
@@ -62,6 +76,7 @@ class ChatIn(BaseModel):
class StartSubagentIn(BaseModel): class StartSubagentIn(BaseModel):
prompt: str prompt: str
llm: Any = None
class ApprovalIn(BaseModel): class ApprovalIn(BaseModel):
prompt: str prompt: str
@@ -70,6 +85,7 @@ class ApprovalIn(BaseModel):
class SubagentActionIn(BaseModel): class SubagentActionIn(BaseModel):
action: str = "intervene" # intervene | abort | kill action: str = "intervene" # intervene | abort | kill
msg: str = "" msg: str = ""
llm: Any = None
@dataclass @dataclass
class SubAgentState: class SubAgentState:
@@ -215,13 +231,13 @@ class SubagentPool:
s.agent.task_queue.put("EXIT") s.agent.task_queue.put("EXIT")
with self.lock: self.subagents.pop(sid, None) with self.lock: self.subagents.pop(sid, None)
if to_abort: push_cards() if to_abort: push_cards()
def start_subagent(self, prompt: str) -> dict: def start_subagent(self, prompt: str, llm: Any = None) -> dict:
sid = short_id() sid = short_id()
agent = GenericAgent() agent = GenericAgent()
agent.inc_out = True agent.inc_out = True
agent.verbose = False agent.verbose = False
agent.no_print = True agent.no_print = True
_apply_desktop_model(agent) if not _select_llm(agent, llm): _apply_desktop_model(agent)
th = start_agent_runner(agent, f"subagent-{sid}") th = start_agent_runner(agent, f"subagent-{sid}")
state = SubAgentState(id=sid, agent=agent, prompt=prompt, status="running", thread=th) state = SubAgentState(id=sid, agent=agent, prompt=prompt, status="running", thread=th)
with self.lock: self.subagents[sid] = state with self.lock: self.subagents[sid] = state
@@ -233,10 +249,11 @@ class SubagentPool:
threading.Thread(target=monitor_display_queue, args=(sid, dq, True), name=f"monitor-{sid}", daemon=True).start() threading.Thread(target=monitor_display_queue, args=(sid, dq, True), name=f"monitor-{sid}", daemon=True).start()
push_cards() push_cards()
return {"id": sid, "status": "running"} return {"id": sid, "status": "running"}
def input_subagent(self, sid: str, msg: str) -> dict: def input_subagent(self, sid: str, msg: str, llm: Any = None) -> dict:
with self.lock: s = self.subagents.get(sid) with self.lock: s = self.subagents.get(sid)
if not s: return {"error": "subagent not found", "id": 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} if s.status == "running": return {"error": "subagent is still running, cannot input/reply. Start a new subagent instead.", "id": sid}
_select_llm(s.agent, llm)
s.prompt = msg s.prompt = msg
s.reply = "" s.reply = ""
s.status = "running" s.status = "running"
@@ -257,10 +274,10 @@ READMES = {
Conductor API\tBase: http://{HOST}:{PORT} Conductor API\tBase: http://{HOST}:{PORT}
POST /chat\tbody: {{"msg": "..."}}\t给用户发消息 POST /chat\tbody: {{"msg": "..."}}\t给用户发消息
POST /subagent\tbody: {{"prompt": "..."}}\t启动新subagent,返回 {{"id": "xxx"}} POST /subagent\tbody: {{"prompt": "..."}}\t启动新subagent,返回 {{"id": "xxx"}};指定模型加参数llm(数字/名称)
POST /approval\tbody: {{"prompt": "...", "source": "..."}}\t推一条待批任务到前端(后端不存),用户同意则直接派发为subagent POST /approval\tbody: {{"prompt": "...", "source": "..."}}\t推一条待批任务到前端(后端不存),用户同意则直接派发为subagent
POST /subagent/{{id}}\tbody: {{"action": "keyinfo", "msg": "..."}}\t注入key_info(agent下轮可见) 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": "input", "msg": "..."}}\t开新一轮任务;指定模型加参数llm(数字/名称)
POST /subagent/{{id}}\tbody: {{"action": "stop"}}\t中断执行但保留(可继续input/reply) POST /subagent/{{id}}\tbody: {{"action": "stop"}}\t中断执行但保留(可继续input/reply)
GET /chat?last=N\t返回最近N条对话(默认20) GET /chat?last=N\t返回最近N条对话(默认20)
GET /subagent\t返回 {{"items": [...]}}\t查看所有subagent状态 GET /subagent\t返回 {{"items": [...]}}\t查看所有subagent状态
@@ -452,7 +469,7 @@ INSTR_DISPATCHED = "Task received. I'll handle THIS TASK from here. You MUST to
@app.post("/subagent") @app.post("/subagent")
def api_start_subagent(body: StartSubagentIn): def api_start_subagent(body: StartSubagentIn):
result = pool.start_subagent(body.prompt) result = pool.start_subagent(body.prompt, body.llm)
result["instruction"] = INSTR_DISPATCHED result["instruction"] = INSTR_DISPATCHED
return result return result
@@ -466,7 +483,7 @@ def api_subagent_action(sid: str, body: SubagentActionIn):
result["instruction"] = "Received. I'll incorporate this. You MUST to do other task or end your reply." result["instruction"] = "Received. I'll incorporate this. You MUST to do other task or end your reply."
return result return result
if action in ("input", "reply", "append", "message", "msg"): if action in ("input", "reply", "append", "message", "msg"):
result = pool.input_subagent(sid, body.msg) result = pool.input_subagent(sid, body.msg, body.llm)
result["instruction"] = INSTR_DISPATCHED result["instruction"] = INSTR_DISPATCHED
return result return result
if action in ("abort", "stop"): if action in ("abort", "stop"):
@@ -479,10 +496,11 @@ def api_subagent_action(sid: str, body: SubagentActionIn):
@app.get("/chat") @app.get("/chat")
def api_get_chat(last: int = 20): def api_get_chat(last: int = 20):
items = [m.copy() for m in chat_messages[-last:]]
for m in chat_messages: for m in chat_messages:
if m.get("role") == "user" and not m.get("read"): m["read"] = True if m.get("role") == "user" and not m.get("read"): m["read"] = True
schedule_broadcast({"type": "chat_read"}) schedule_broadcast({"type": "chat_read"})
return {"items": chat_messages[-last:]} return {"items": items}
@app.post("/chat") @app.post("/chat")
def api_chat(body: ChatIn): def api_chat(body: ChatIn):
+10 -6
View File
@@ -1081,10 +1081,9 @@ def render_native_history(history, ts=None):
`_parse_native_history(_pairs(render_native_history(h))) == h`. `_parse_native_history(_pairs(render_native_history(h))) == h`.
Byte-for-byte matches llmcore's `_write_llm_log` for the tool backend: Byte-for-byte matches llmcore's `_write_llm_log` for the tool backend:
Prompt = `json.dumps(merged, ensure_ascii=False, indent=2)`; Response = Prompt/Response each put one content block per line, with their wrappers
`str(content_blocks)` which for a list equals `repr(...)` (survives attached to the first/last lines. Response keeps Python repr so
True/False/None that JSON would emit as true/false/null and break True/False/None survive `ast.literal_eval`. Only complete pairs are emitted."""
`ast.literal_eval`). Only complete (user, assistant) pairs are emitted."""
ts = ts or datetime.now().strftime('%Y-%m-%d %H:%M:%S') ts = ts or datetime.now().strftime('%Y-%m-%d %H:%M:%S')
out, i, n = [], 0, len(history) out, i, n = [], 0, len(history)
while i < n: while i < n:
@@ -1092,8 +1091,13 @@ def render_native_history(history, ts=None):
if isinstance(u, dict) and u.get('role') == 'user' and i + 1 < n: if isinstance(u, dict) and u.get('role') == 'user' and i + 1 < n:
a = history[i + 1] a = history[i + 1]
if isinstance(a, dict) and a.get('role') == 'assistant': if isinstance(a, dict) and a.get('role') == 'assistant':
out.append(f"=== Prompt === {ts}\n{json.dumps(u, ensure_ascii=False, indent=2)}\n\n") prompt_blocks = u.get('content', [])
out.append(f"=== Response === {ts}\n{a.get('content')!r}\n\n") prompt_raw = '{"role": "user", "content": [\n' + ",\n".join(
json.dumps(b, ensure_ascii=False) for b in prompt_blocks) + "]}"
out.append(f"=== Prompt === {ts}\n{prompt_raw}\n\n")
blocks = a.get('content', [])
raw = "[" + ",\n".join(repr(b) for b in blocks) + "]"
out.append(f"=== Response === {ts}\n{raw}\n\n")
i += 2 i += 2
continue continue
i += 1 i += 1
+8 -3
View File
@@ -557,6 +557,7 @@ class BaseSession:
self.service_tier = _enum('service_tier', {'auto', 'default', 'priority', 'flex'}) self.service_tier = _enum('service_tier', {'auto', 'default', 'priority', 'flex'})
self.thinking_type = _enum('thinking_type', {'adaptive', 'enabled', 'disabled'}) self.thinking_type = _enum('thinking_type', {'adaptive', 'enabled', 'disabled'})
self.thinking_budget_tokens = cfg.get('thinking_budget_tokens') self.thinking_budget_tokens = cfg.get('thinking_budget_tokens')
self.omit_thinking = cfg.get('omit_thinking', False) # Exclude thinking from session history
mode = str(cfg.get('api_mode', 'chat_completions')).strip().lower().replace('-', '_') mode = str(cfg.get('api_mode', 'chat_completions')).strip().lower().replace('-', '_')
self.api_mode = 'responses' if mode in ('responses', 'response') else 'chat_completions' self.api_mode = 'responses' if mode in ('responses', 'response') else 'chat_completions'
self.temperature = cfg.get('temperature', 1) self.temperature = cfg.get('temperature', 1)
@@ -735,7 +736,9 @@ class NativeClaudeSession(BaseSession):
except StopIteration as e: content_blocks = e.value or [] except StopIteration as e: content_blocks = e.value or []
if content_blocks and (_injected := _ensure_text_block(content_blocks)): yield _injected if content_blocks and (_injected := _ensure_text_block(content_blocks)): yield _injected
if content_blocks and not (len(content_blocks) == 1 and content_blocks[0].get("text", "").startswith("!!!Error:")): if content_blocks and not (len(content_blocks) == 1 and content_blocks[0].get("text", "").startswith("!!!Error:")):
self.history.append({"role": "assistant", "content": content_blocks}) history_blocks = content_blocks
if self.omit_thinking: history_blocks = [b for b in content_blocks if b.get("type") != "thinking"]
self.history.append({"role": "assistant", "content": history_blocks})
text_parts = [b["text"] for b in content_blocks if b.get("type") == "text"] text_parts = [b["text"] for b in content_blocks if b.get("type") == "text"]
content = "\n".join(text_parts).strip() content = "\n".join(text_parts).strip()
tool_calls = [MockToolCall(b["name"], b.get("input", {}), id=b.get("id", "")) for b in content_blocks if b.get("type") == "tool_use"] tool_calls = [MockToolCall(b["name"], b.get("input", {}), id=b.get("id", "")) for b in content_blocks if b.get("type") == "tool_use"]
@@ -748,7 +751,8 @@ class NativeClaudeSession(BaseSession):
if think_match: if think_match:
thinking = think_match.group(1).strip() thinking = think_match.group(1).strip()
content = re.sub(think_pattern, "", content, flags=re.DOTALL) content = re.sub(think_pattern, "", content, flags=re.DOTALL)
return MockResponse(thinking, content, tool_calls, str(content_blocks)) raw = "[" + ",\n".join(repr(b) for b in content_blocks) + "]"
return MockResponse(thinking, content, tool_calls, raw)
class NativeOAISession(NativeClaudeSession): class NativeOAISession(NativeClaudeSession):
native_ua = "codex_exec/0.139.0 (Windows 10.0.26200; x86_64) unknown (codex_exec; 0.139.0)" native_ua = "codex_exec/0.139.0 (Windows 10.0.26200; x86_64) unknown (codex_exec; 0.139.0)"
@@ -1060,7 +1064,8 @@ class NativeToolClient:
final_content = tool_result_blocks + filtered_content final_content = tool_result_blocks + filtered_content
if not final_content: final_content = [{"type": "text", "text": "."}] if not final_content: final_content = [{"type": "text", "text": "."}]
merged = {"role": "user", "content": final_content} merged = {"role": "user", "content": final_content}
_write_llm_log('Prompt', json.dumps(merged, ensure_ascii=False, indent=2), self.log_path) prompt_raw = '{"role": "user", "content": [\n' + ",\n".join(json.dumps(b, ensure_ascii=False) for b in final_content) + "]}"
_write_llm_log('Prompt', prompt_raw, self.log_path)
gen = self.backend.ask(merged) gen = self.backend.ask(merged)
try: try:
while True: while True: