"""智能体服务（AgentService）—— zcode 式"模型操作工作区文件"的网关侧封装。 职责： - OpenAICompatChat：OpenAI 兼容 /chat/completions 的工具调用客户端（ToolLoop 的 chat_fn）， 支持 httpx transport/client 注入（测试用 MockTransport，对齐 D11 封闭性）。 - AgentService：运行一次智能体任务——事件逐条落盘 agent_runs/{id}/events.jsonl， 终态写 status.json；SSE 端点轮询事件文件增量推送（与 v3 workspace 监视同思路， 不侵入 router_system）。 - 模型来源：模型池 agent 角色（或显式 pool_id），否则回退经典 Architect 设置。 安全与护栏： - 文件操作被 WorkspaceTools 关押在工作区根目录内 - 轮数上限（agent.max_rounds）与 token 熔断（agent.token_cap）双护栏 """ from __future__ import annotations import asyncio import json import time import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Any, Awaitable, Callable, Dict, List, Optional from router_system.tools import ToolLoop, WorkspaceTools # 运行目录（与 runs/ 平级） AGENT_RUNS_DIR = Path("agent_runs") STATE_RUNNING = "running" STATE_DONE = "done" STATE_FAILED = "failed" AGENT_SYSTEM_PROMPT = ( "你是端云协同 LLM 系统中的智能体（Agent），正在操作用户选择的**真实项目工作目录**。" "你拥有的工具：list_dir（列目录）、read_file（读文件）、write_file（写文件/新建）、" "edit_file（精确替换编辑：old_string 须唯一匹配）、search_files（跨文件搜索内容）、" "run_command（执行 shell 命令，仅当系统开启 allow_shell 时可用，否则不要尝试）、" "web_fetch（抓取公网 http/https 文档页面，私网地址会被拒绝）。" "像编程助手一样工作：先列目录/搜索了解项目结构，读文件核对原文后再用 edit_file 小步修改" "（或 write_file 新建），需要查外部资料时用 web_fetch，需要时运行命令验证。" "任务完成或给出结论后，直接输出给用户的最终答复（中文，不要再调用工具）。" "注意：不要反复以完全相同的参数调用同一工具——那不会带来新信息。" ) # ── 两级智能体（D7）：规划者（大模型）+ 执行者（本地小模型），交接走 handoff 文档 ── PLANNER_SYSTEM_PROMPT = ( "你是两级智能体中的**规划者**（大模型）。执行者是一个能力有限的本地小模型，" "只能机械地使用工具。你的职责：把用户任务拆成执行者可照做的**具体指令**，" "并在执行后审查其汇报。输出必须是合法 JSON 对象（不要 markdown 围栏）。" ) EXECUTOR_SYSTEM_PROMPT = ( "你是两级智能体中的**执行者**（本地小模型）。规划者已给你具体指令，" "你只负责用工具完成指令并在最后**汇报**：做了什么、结果如何、有什么问题。" "严格遵守指令范围，不要自行扩大任务。汇报用中文，是给规划者看的，" "要列出：修改的文件、关键命令输出、未完成项。" ) # 规划者首轮：产出指令（JSON） _PLAN_SCHEMA_HINT = { "instructions": "string(给执行者的具体步骤指令，<=600字)", "acceptance": "string(验收标准，<=200字)", } # 规划者审查轮：裁决（JSON） _REVIEW_SCHEMA_HINT = { "verdict": "enum(done|redo)", "reply_to_executor": "string(verdict=redo 时给执行者的补充指令；done 时可空)", "final_answer": "string(verdict=done 时给用户的最终答复)", } DEFAULT_MAX_HANDOFFS = 2 # 规划者<->执行者交接轮数上限 def _parse_json_loose(content: str) -> Dict[str, Any]: """宽松解析规划者的 JSON 输出（剥围栏/取首个对象）；失败返回 {}。""" try: from router_system.architect import ArchitectClient return ArchitectClient._parse_json(content) except Exception: return {} # ───────────────────────────────────────────────────────────────────────────── # OpenAI 兼容工具调用客户端 # ───────────────────────────────────────────────────────────────────────────── class OpenAICompatChat: """ToolLoop.chat_fn 的 OpenAI 兼容实现（支持 tools 参数）。""" def __init__( self, base_url: str, api_key: Optional[str], model: str, temperature: float = 0.3, max_tokens: int = 4096, timeout_s: float = 120.0, stream: bool = True, max_retries: int = 2, retry_delay_s: float = 1.0, transport: Any = None, _client: Any = None, ): self.base_url = base_url.rstrip("/") self.api_key = api_key self.model = model self.temperature = temperature self.max_tokens = max_tokens self.timeout_s = timeout_s self.stream = stream # D10：默认流式；解析失败自动回退非流式 self.max_retries = max(0, int(max_retries)) # 可重试错误的重试次数（dsh llm-retry 同款） self.retry_delay_s = max(0.0, float(retry_delay_s)) # 指数退避基数 self._transport = transport self._client = _client self._owns = _client is None def _get_client(self): if self._client is None: import httpx kwargs: Dict[str, Any] = {"timeout": self.timeout_s} if self._transport is not None: kwargs["transport"] = self._transport self._client = httpx.AsyncClient(**kwargs) return self._client async def aclose(self) -> None: if self._owns and self._client is not None: await self._client.aclose() self._client = None async def __call__(self, messages: List[Dict[str, Any]], tools_spec: List[Dict[str, Any]], on_delta: Optional[Callable[[str], None]] = None) -> Dict[str, Any]: """ToolLoop.chat_fn：默认流式（D10）；流式不可用时回退非流式（带重试退避）。""" if self.stream: try: return await self._stream_call(messages, tools_spec, on_delta) except Exception: # 已有部分增量输出则如实抛出；否则回退非流式 if getattr(self, "_stream_partial", False): raise return await self._post_with_retry(messages, tools_spec) @staticmethod def _is_retryable(exc: Exception) -> bool: """可重试错误：网络传输类 / 408 / 429 / 5xx（dsh retryableCodes 同思路）。""" import httpx if isinstance(exc, httpx.TransportError): return True if isinstance(exc, httpx.HTTPStatusError): code = exc.response.status_code return code in (408, 429) or code >= 500 return False async def _post_with_retry(self, messages: List[Dict[str, Any]], tools_spec: List[Dict[str, Any]]) -> Dict[str, Any]: """非流式调用 + 指数退避重试（仅针对可重试错误）。""" for attempt in range(self.max_retries + 1): try: return await self._post_once(messages, tools_spec) except Exception as exc: if attempt >= self.max_retries or not self._is_retryable(exc): raise await asyncio.sleep(self.retry_delay_s * (2 ** attempt)) async def _stream_call(self, messages: List[Dict[str, Any]], tools_spec: List[Dict[str, Any]], on_delta: Optional[Callable[[str], None]]) -> Dict[str, Any]: """流式调用：逐段转发 content 增量；tool_calls 碎片按 index 组装（不在正文展示）。""" import json as _json self._stream_partial = False # 每次调用前复位（防上次的标志影响本次回退判定） body: Dict[str, Any] = { "model": self.model, "messages": messages, "temperature": self.temperature, "max_tokens": self.max_tokens, "stream": True, "stream_options": {"include_usage": True}, } if tools_spec: body["tools"] = tools_spec body["tool_choice"] = "auto" headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {} client = self._get_client() content_parts: List[str] = [] tc_slots: Dict[int, Dict[str, str]] = {} usage: Dict[str, Any] = {} async with client.stream("POST", f"{self.base_url}/chat/completions", headers=headers, json=body) as resp: resp.raise_for_status() async for line in resp.aiter_lines(): line = line.strip() if not line.startswith("data:"): continue payload = line[5:].strip() if payload == "[DONE]": break try: obj = _json.loads(payload) except _json.JSONDecodeError: continue choices = obj.get("choices") or [{}] delta = (choices[0].get("delta") or {}) if choices else {} piece = delta.get("content") if piece: self._stream_partial = True content_parts.append(piece) if on_delta is not None: try: on_delta(piece) except Exception: pass for tc in delta.get("tool_calls") or []: idx = int(tc.get("index", 0)) slot = tc_slots.setdefault(idx, {"id": "", "name": "", "args": ""}) if tc.get("id"): slot["id"] = tc["id"] fn = tc.get("function") or {} if fn.get("name"): slot["name"] = fn["name"] if fn.get("arguments"): slot["args"] += fn["arguments"] if obj.get("usage"): usage = obj["usage"] content = "".join(content_parts) or None from router_system.tools import _loads_json_object tool_calls = [] for idx in sorted(tc_slots): slot = tc_slots[idx] tool_calls.append({ "id": slot["id"] or f"call_{idx}", "name": slot["name"], "arguments": _loads_json_object(slot["args"]), }) return {"content": content, "tool_calls": tool_calls, "usage": usage} async def _post_once(self, messages: List[Dict[str, Any]], tools_spec: List[Dict[str, Any]]) -> Dict[str, Any]: """非流式调用（回退路径）。""" body: Dict[str, Any] = { "model": self.model, "messages": messages, "temperature": self.temperature, "max_tokens": self.max_tokens, } if tools_spec: body["tools"] = tools_spec body["tool_choice"] = "auto" headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {} client = self._get_client() resp = await client.post(f"{self.base_url}/chat/completions", headers=headers, json=body) resp.raise_for_status() data = resp.json() msg = (data.get("choices") or [{}])[0].get("message") or {} from router_system.tools import parse_tool_calls return { "content": msg.get("content"), "tool_calls": parse_tool_calls(msg), "usage": data.get("usage") or {}, } # ───────────────────────────────────────────────────────────────────────────── # 智能体服务 # ───────────────────────────────────────────────────────────────────────────── @dataclass class AgentRunInfo: """一次智能体运行的状态快照（内存 + status.json 双写）。""" request_id: str task: str = "" model: str = "" state: str = STATE_RUNNING started_at: float = 0.0 finished_at: float = 0.0 error: Optional[str] = None response: str = "" rounds: int = 0 prompt_tokens: int = 0 completion_tokens: int = 0 pool_id: str = "" workspace: str = "" # 本次运行使用的工作区根目录（绝对路径） executor_model: str = "" # 两级模式：执行者模型名（空 = 单模型模式） mode: str = "single" # single | dual tool_calls: int = 0 # 本次运行的工具调用步数 asyncio_task: Optional[asyncio.Task] = field(default=None, repr=False) def to_dict(self) -> Dict[str, Any]: return { "request_id": self.request_id, "task": self.task, "model": self.model, "state": self.state, "started_at": self.started_at, "finished_at": self.finished_at, "error": self.error, "response": self.response, "rounds": self.rounds, "prompt_tokens": self.prompt_tokens, "completion_tokens": self.completion_tokens, "pool_id": self.pool_id, "workspace": self.workspace, "executor_model": self.executor_model, "mode": self.mode, "tool_calls": self.tool_calls, } class AgentService: """智能体运行服务：事件落盘 + 状态管理。""" def __init__(self, run_dir: str | Path = AGENT_RUNS_DIR): self.run_dir = Path(run_dir) self._runs: Dict[str, AgentRunInfo] = {} self.max_running = 5 # ---------- 路径 ---------- def _dir(self, request_id: str) -> Path: return self.run_dir / request_id def events_path(self, request_id: str) -> Path: return self._dir(request_id) / "events.jsonl" def status_path(self, request_id: str) -> Path: return self._dir(request_id) / "status.json" # ---------- 注册与查询 ---------- def register(self, request_id: str, task: str, model: str, pool_id: str, workspace: str = "", executor_model: str = "", mode: str = "single") -> Optional[AgentRunInfo]: running = [r for r in self._runs.values() if r.state == STATE_RUNNING] if len(running) >= self.max_running: return None info = AgentRunInfo(request_id=request_id, task=task, model=model, pool_id=pool_id, workspace=workspace, executor_model=executor_model, mode=mode, started_at=time.time()) self._runs[request_id] = info self._dir(request_id).mkdir(parents=True, exist_ok=True) self._write_status(info) return info def get(self, request_id: str) -> Optional[AgentRunInfo]: return self._runs.get(request_id) # ---------- 执行 ---------- async def run(self, info: AgentRunInfo, chat: Any, workspace_dir: str | Path, max_rounds: int = 8, token_cap: int = 0, allow_shell: bool = False, shell_timeout_s: int = 20, allow_net: bool = True, executor_chat: Any = None, max_handoffs: int = DEFAULT_MAX_HANDOFFS, session: Optional["AgentSession"] = None, approval_policy: str = "dangerous", approval_timeout_s: int = 120) -> None: """执行智能体任务（由调用方包成后台协程）。 executor_chat 为空 = 单模型模式（chat 全程包办）； 提供时进入两级模式：chat 作规划者，executor_chat 作执行者（D7）。 session 提供时：既往轮次作为对话上下文，完成后把本轮追加进会话。 """ history = self._history_from_session(session) approval_mgr = ApprovalManager() info._approval_manager = approval_mgr # 供 /approve 端点裁决（瞬态属性） throttle = DeltaThrottle(lambda ev: self._append_event(info, ev)) async def approval_hook(name: str, args: Dict[str, Any]) -> bool: """按策略判定；需审批则挂起等用户裁决，超时 fail-closed。""" if not needs_approval(approval_policy, name): return True aid = "ap" + uuid.uuid4().hex[:8] ev = approval_mgr.open(aid) self._append_event(info, {"type": "approval_request", "id": aid, "name": name, "arguments": args, "policy": approval_policy}) # 轮询等待（0.1s 步进）：不用 wait_for——portal 循环下其定时器不可靠 allowed = False note = "" deadline = time.time() + max(1, approval_timeout_s) while time.time() < deadline: if ev.is_set(): allowed = approval_mgr._pending.get(aid, {}).get("allowed", False) break await asyncio.sleep(0.1) else: note = f"超时（{approval_timeout_s}s）未响应，自动拒绝" if ev.is_set() and not allowed: note = note or "用户拒绝" approval_mgr.close(aid) self._append_event(info, {"type": "approval_decided", "id": aid, "name": name, "allowed": allowed, **({"note": note} if note else {})}) return allowed try: if executor_chat is not None: result = await self.run_dual( info, chat, executor_chat, workspace_dir, max_rounds=max_rounds, token_cap=token_cap, allow_shell=allow_shell, shell_timeout_s=shell_timeout_s, allow_net=allow_net, max_handoffs=max_handoffs, approval_hook=approval_hook) else: tools = WorkspaceTools(workspace_dir, allow_shell=allow_shell, shell_timeout_s=shell_timeout_s, allow_net=allow_net) loop = ToolLoop(tools, chat, max_rounds=max_rounds, token_cap=token_cap, on_event=self._make_event_writer(info), approval_hook=approval_hook, on_delta=throttle.make_cb("executor")) result = await loop.run(info.task, system=AGENT_SYSTEM_PROMPT, history=history) throttle.flush("executor") self._apply_result(info, result) except Exception as exc: # pragma: no cover info.state = STATE_FAILED info.error = f"{type(exc).__name__}: {exc}" self._append_event(info, {"type": "final", "round": info.rounds, "reason": "error", "error": info.error}) finally: info.finished_at = time.time() self._write_status(info) if session is not None: session.data["turns"].append({ "request_id": info.request_id, "task": info.task, "response": info.response, "state": info.state, "tool_calls": info.tool_calls, "tokens": info.prompt_tokens + info.completion_tokens, "error": info.error, "ts": info.finished_at, }) get_session_store().save(session) @staticmethod def _history_from_session(session: Optional["AgentSession"], max_turns: int = 6, max_chars: int = 1500) -> List[Dict[str, Any]]: """把会话既往轮次折叠成对话上下文（不含工具细节）。""" if session is None: return [] turns = [t for t in session.data.get("turns", []) if t.get("state") == STATE_DONE and t.get("response")] out: List[Dict[str, Any]] = [] for t in turns[-max_turns:]: out.append({"role": "user", "content": str(t["task"])[:max_chars]}) out.append({"role": "assistant", "content": str(t["response"])[:max_chars]}) return out def _apply_result(self, info: AgentRunInfo, result: Dict[str, Any]) -> None: """把循环结果落到运行状态（单/两级模式共用）。""" info.response = result.get("response", "") info.rounds = int(result.get("rounds", 0)) info.prompt_tokens = int(result.get("prompt_tokens", 0)) info.completion_tokens = int(result.get("completion_tokens", 0)) if result.get("reason") == "error": info.state = STATE_FAILED info.error = result.get("error") elif result.get("reason") in ("token_cap", "max_rounds", "max_handoffs"): # 触顶属于护栏行为：结果仍交付，但标记部分完成信息 info.state = STATE_DONE info.error = result.get("error") else: info.state = STATE_DONE # ---------- 两级模式（D7）：规划者 + 执行者 ---------- async def run_dual(self, info: AgentRunInfo, planner_chat: Any, executor_chat: Any, workspace_dir: str | Path, max_rounds: int = 8, token_cap: int = 0, allow_shell: bool = False, shell_timeout_s: int = 20, allow_net: bool = True, max_handoffs: int = DEFAULT_MAX_HANDOFFS, approval_hook: Optional[Callable[[str, Dict[str, Any]], Awaitable[bool]]] = None ) -> Dict[str, Any]: """大模型拆解/审查 + 小模型执行工具轮，交接状态写 handoff.json（智能体版交流文本）。""" info.mode = "dual" tools = WorkspaceTools(workspace_dir, allow_shell=allow_shell, shell_timeout_s=shell_timeout_s, allow_net=allow_net) throttle = DeltaThrottle(lambda ev: self._append_event(info, ev)) handoff: Dict[str, Any] = { "task": info.task, "planner_model": info.model, "executor_model": info.executor_model, "workspace": info.workspace, "instructions": "", "acceptance": "", "exchanges": [], } spent = {"in": 0, "out": 0} total_rounds = 0 def _account(usage: Dict[str, Any] | None) -> None: spent["in"] += int((usage or {}).get("prompt_tokens", 0)) spent["out"] += int((usage or {}).get("completion_tokens", 0)) def _save_handoff() -> None: try: (self._dir(info.request_id) / "handoff.json").write_text( json.dumps(handoff, ensure_ascii=False, indent=2), encoding="utf-8") except OSError: pass def _remaining_cap() -> int: return (token_cap - spent["in"] - spent["out"]) if token_cap else 1 async def _planner_json(user_msg: str) -> Dict[str, Any]: """调规划者并解析 JSON；解析失败回喂重试一次，再失败降级为 {}（禁止带病继续的软版本）。""" import inspect messages = [{"role": "system", "content": PLANNER_SYSTEM_PROMPT}, {"role": "user", "content": user_msg}] content = "" for attempt in (1, 2): # 规划者同样流式（前端弱化展示其 JSON 草稿） try: accepts = len(inspect.signature(planner_chat).parameters) >= 3 except (TypeError, ValueError): accepts = False if accepts: resp = await planner_chat(messages, [], throttle.make_cb("planner")) else: resp = await planner_chat(messages, []) _account(resp.get("usage")) content = resp.get("content") or "" throttle.flush("planner") obj = _parse_json_loose(content) if obj: break if attempt == 1: messages += [{"role": "assistant", "content": content}, {"role": "user", "content": "你的输出不是合法 JSON。请重新只输出合法 JSON 对象。"}] self._append_event(info, {"type": "message", "role": "planner", "content": content[:2000]}) return obj try: # ---- 阶段 1：规划（大模型拆解为执行者指令） ---- self._append_event(info, {"type": "phase", "phase": "plan", "model": info.model}) plan = await _planner_json( f"用户任务：{info.task}\n\n" "请产出给执行者的指令，仅输出符合如下结构的 JSON：\n" + json.dumps(_PLAN_SCHEMA_HINT, ensure_ascii=False)) instructions = (plan.get("instructions") or info.task).strip() handoff["instructions"] = instructions handoff["acceptance"] = str(plan.get("acceptance", "")) _save_handoff() final_text = "" reason = "answer" error = None exec_rounds_total = 0 # ---- 阶段 2/3：执行 <-> 审查（有界交接） ---- for h in range(1, max_handoffs + 1): # 执行（本地小模型跑工具轮） self._append_event(info, {"type": "phase", "phase": "execute", "handoff": h, "model": info.executor_model}) loop = ToolLoop(tools, executor_chat, max_rounds=max_rounds, token_cap=max(1, _remaining_cap()), on_event=self._make_event_writer(info), emit_final=False, approval_hook=approval_hook, on_delta=throttle.make_cb("executor")) exec_result = await loop.run(instructions, system=EXECUTOR_SYSTEM_PROMPT) _account({"prompt_tokens": exec_result.get("prompt_tokens", 0), "completion_tokens": exec_result.get("completion_tokens", 0)}) exec_rounds_total += int(exec_result.get("rounds", 0)) report = exec_result.get("response", "") # 执行者汇报作为消息事件透出（前端可读） self._append_event(info, {"type": "message", "role": "executor", "handoff": h, "content": (report or "")[:4000]}) if exec_result.get("reason") == "error": reason, error = "error", exec_result.get("error") final_text = report break # 审查（大模型裁决） self._append_event(info, {"type": "phase", "phase": "review", "handoff": h, "model": info.model}) review = await _planner_json( f"用户任务：{info.task}\n你之前给出的指令：{instructions}\n" f"验收标准：{handoff['acceptance'] or '（未明确）'}\n\n" f"执行者第 {h} 轮汇报：\n{report[:4000]}\n\n" "请审查是否已按验收标准完成，仅输出符合如下结构的 JSON：\n" + json.dumps(_REVIEW_SCHEMA_HINT, ensure_ascii=False)) verdict = str(review.get("verdict", "done")).lower() handoff["exchanges"].append({ "handoff": h, "executor_report": report, "verdict": verdict, "reply_to_executor": str(review.get("reply_to_executor", "")), }) _save_handoff() if verdict == "done": final_text = str(review.get("final_answer") or report) break # redo：裁决意见作为下一轮执行者指令（带上一轮上下文） instructions = str(review.get("reply_to_executor") or instructions) if h == max_handoffs: reason = "max_handoffs" error = f"交接轮数达上限（{max_handoffs}），以执行者汇报收尾" final_text = report else: final_text = final_text or "" self._append_event(info, {"type": "final", "round": total_rounds + exec_rounds_total, "reason": reason, "error": error}) return {"response": final_text, "rounds": total_rounds + exec_rounds_total, "reason": reason, "error": error, "prompt_tokens": spent["in"], "completion_tokens": spent["out"]} except Exception as exc: reason = "error" error = f"{type(exc).__name__}: {exc}" self._append_event(info, {"type": "final", "round": total_rounds, "reason": reason, "error": error}) return {"response": "", "rounds": total_rounds, "reason": reason, "error": error, "prompt_tokens": spent["in"], "completion_tokens": spent["out"]} # ---------- 事件 ---------- def _make_event_writer(self, info: AgentRunInfo): def _on_event(ev: Dict[str, Any]) -> None: if ev.get("type") == "tool_call": info.tool_calls += 1 # 工具步数统计（单/两级模式统一在此） self._append_event(info, ev) return _on_event def _append_event(self, info: AgentRunInfo, ev: Dict[str, Any]) -> None: ev = {"ts": time.time(), **ev} try: with self.events_path(info.request_id).open("a", encoding="utf-8") as f: f.write(json.dumps(ev, ensure_ascii=False) + "\n") except OSError: pass def read_events(self, request_id: str) -> List[Dict[str, Any]]: p = self.events_path(request_id) if not p.exists(): return [] out = [] for line in p.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line: continue try: out.append(json.loads(line)) except json.JSONDecodeError: pass # 半行（正在写入）忽略 return out # ---------- 状态 ---------- def _write_status(self, info: AgentRunInfo) -> None: try: self.status_path(info.request_id).write_text( json.dumps(info.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8") except OSError: pass async def watch_events(self, request_id: str, cancel_event: asyncio.Event, poll_interval: float = 0.3, max_seconds: float = 900.0): """SSE 生成器：增量推送 events.jsonl 新行，直到终态/取消/超时。 从文件头开始回放（晚加入的订阅者也能看到完整过程）。 """ p = self.events_path(request_id) offset = 0 deadline = time.time() + max_seconds while not cancel_event.is_set() and time.time() < deadline: if p.exists(): try: size = p.stat().st_size if size > offset: with p.open("r", encoding="utf-8") as f: f.seek(offset) new_text = f.read() offset = f.tell() for line in new_text.splitlines(): line = line.strip() if not line: continue try: ev = json.loads(line) except json.JSONDecodeError: continue yield ev if ev.get("type") == "final": return except OSError: pass info = self.get(request_id) if info and info.state in (STATE_DONE, STATE_FAILED): # 终态兜底：状态已结束但可能没有 final 事件（如注册即失败） yield {"type": "final", "round": info.rounds, "reason": "answer" if info.state == STATE_DONE else "error", "error": info.error} return await asyncio.sleep(poll_interval) yield {"type": "final", "round": 0, "reason": "error", "error": "订阅超时"} # ---------- 全局单例 ---------- _service: Optional[AgentService] = None def get_agent_service() -> AgentService: global _service if _service is None: _service = AgentService() return _service def reset_agent_service() -> None: """测试用：重置全局智能体服务单例。""" global _service _service = None def new_request_id() -> str: return "ag" + uuid.uuid4().hex[:10] # ───────────────────────────────────────────────────────────────────────────── # 审批流（D9）：dsh 式 allow-once / deny，fail-closed # ───────────────────────────────────────────────────────────────────────────── READ_ONLY_TOOLS = {"list_dir", "read_file", "search_files", "web_fetch"} def needs_approval(policy: str, tool_name: str) -> bool: """审批策略判定：off=全放行；all=全询问；dangerous=写/编辑/命令询问，只读放行。""" if policy == "all": return True if policy == "dangerous": return tool_name not in READ_ONLY_TOOLS return False class DeltaThrottle: """流式增量节流（D10）：积攒超过阈值才落一条 delta 事件，防事件爆炸。""" THRESHOLD = 48 def __init__(self, append_event): self._append = append_event # (ev: dict) -> None self._buf: Dict[str, str] = {} def make_cb(self, role: str): def cb(text: str) -> None: self.add(role, text) return cb def add(self, role: str, text: str) -> None: buf = self._buf.get(role, "") + (text or "") if len(buf) >= self.THRESHOLD: self._flush(role, buf) buf = "" self._buf[role] = buf def flush(self, role: Optional[str] = None) -> None: roles = [role] if role else list(self._buf.keys()) for r in roles: buf = self._buf.get(r, "") if buf: self._flush(r, buf) self._buf[r] = "" def _flush(self, role: str, text: str) -> None: self._append({"type": "delta", "role": role, "text": text}) class ApprovalManager: """单次智能体运行内的审批挂起/裁决（asyncio Event 实现，dsh 式 allow-once）。""" def __init__(self): self._pending: Dict[str, Dict[str, Any]] = {} def open(self, approval_id: str) -> asyncio.Event: ev = asyncio.Event() self._pending[approval_id] = {"event": ev, "allowed": False} return ev def decide(self, approval_id: str, allowed: bool) -> bool: p = self._pending.get(approval_id) if p is None: return False p["allowed"] = allowed p["event"].set() return True def close(self, approval_id: str) -> None: self._pending.pop(approval_id, None) # ───────────────────────────────────────────────────────────────────────────── # 会话（dsh 式：工作区内多轮对话，持久化到磁盘） # ───────────────────────────────────────────────────────────────────────────── SESSIONS_DIR = Path("agent_runs") / "sessions" class AgentSession: """一个智能体会话：多轮任务 + 配置快照（磁盘持久化）。""" def __init__(self, data: Dict[str, Any]): self.data = data @classmethod def new(cls, sid: str, title: str, workspace: str, pool_id: str = "", executor_pool_id: str = "") -> "AgentSession": now = time.time() return cls({ "id": sid, "title": title[:24] or "新会话", "workspace": workspace, "pool_id": pool_id, "executor_pool_id": executor_pool_id, "created_at": now, "updated_at": now, "busy": False, "turns": [], # [{request_id, task, response, state, tool_calls, tokens}] }) def to_dict(self) -> Dict[str, Any]: return dict(self.data) def view(self, include_turns: bool = True) -> Dict[str, Any]: out = self.to_dict() if not include_turns: out["turns"] = len(self.data.get("turns", [])) return out class SessionStore: """会话注册表（内存索引 + sessions/{sid}.json 持久化）。""" def __init__(self, root: Path = SESSIONS_DIR): self.root = Path(root) self.root.mkdir(parents=True, exist_ok=True) self._cache: Dict[str, AgentSession] = {} def _path(self, sid: str) -> Path: return self.root / f"{sid}.json" def create(self, title: str, workspace: str, pool_id: str = "", executor_pool_id: str = "") -> AgentSession: sid = "as" + uuid.uuid4().hex[:10] sess = AgentSession.new(sid, title or "新会话", workspace, pool_id, executor_pool_id) self._cache[sid] = sess self._save(sess) return sess def get(self, sid: str) -> Optional[AgentSession]: if sid in self._cache: return self._cache[sid] p = self._path(sid) if not p.exists(): return None try: sess = AgentSession(json.loads(p.read_text(encoding="utf-8"))) self._cache[sid] = sess return sess except (json.JSONDecodeError, OSError): return None def list(self) -> List[Dict[str, Any]]: out = [] for p in sorted(self.root.glob("*.json"), key=lambda x: x.stat().st_mtime, reverse=True): try: out.append(json.loads(p.read_text(encoding="utf-8"))) except (json.JSONDecodeError, OSError): continue return out def delete(self, sid: str) -> bool: self._cache.pop(sid, None) p = self._path(sid) if p.exists(): p.unlink() return True return False def rename(self, sid: str, title: str) -> Optional[AgentSession]: """重命名会话标题（dsh session.rename 对齐）。""" sess = self.get(sid) if sess is None: return None title = (title or "").strip() if not title: return sess sess.data["title"] = title[:24] self.save(sess) return sess def save(self, sess: AgentSession) -> None: self._cache[sess.data["id"]] = sess self._save(sess) def _save(self, sess: AgentSession) -> None: sess.data["updated_at"] = time.time() try: self._path(sess.data["id"]).write_text( json.dumps(sess.data, ensure_ascii=False, indent=2), encoding="utf-8") except OSError: pass _session_store: Optional[SessionStore] = None def get_session_store() -> SessionStore: global _session_store if _session_store is None: _session_store = SessionStore() return _session_store def reset_session_store() -> None: """测试用。""" global _session_store _session_store = None