"""智能体服务（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, 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（写文件）。" "像程序员助手一样工作：先列目录/读文件了解现状，需要时再写文件；" "任务完成或给出结论后，直接输出给用户的最终答复（中文，不要再调用工具）。" ) # ───────────────────────────────────────────────────────────────────────────── # 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, 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._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]]) -> 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 {} # tool_calls 解析放这里（网关层），内核 tools.parse_tool_calls 供其他调用方复用 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 = "" 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, } 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) -> 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, 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) -> None: """执行智能体任务（由调用方包成后台协程）。""" tools = WorkspaceTools(workspace_dir) loop = ToolLoop(tools, chat, max_rounds=max_rounds, token_cap=token_cap, on_event=self._make_event_writer(info)) try: result = await loop.run(info.task, system=AGENT_SYSTEM_PROMPT) 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"): # 触顶属于护栏行为：结果仍交付，但标记部分完成信息 info.state = STATE_DONE info.error = result.get("error") else: info.state = STATE_DONE 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) # ---------- 事件 ---------- def _make_event_writer(self, info: AgentRunInfo): def _on_event(ev: Dict[str, Any]) -> None: 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]