"""Agent 编排层：OpenAI 兼容模型（DeepSeek 等）经 MCP 工具控制手机。 支持两种运行模式： - run_stream()：流式（SSE 逐 token + 工具调用实时回调）——Web AI 控制台用 - run()：非流式收集结果——CLI 用（内部调 run_stream） 流式细节（OpenAI 兼容）： - content/reasoning_content 增量逐 chunk 回调（kind 区分） - tool_calls 分片累积（arguments 按 index 拼接），流结束后统一执行 - 截图（de_screenshot）图像转 image_url 追加下一轮，同时 on_tool 回调带缩略 - token 用量：请求带 stream_options.include_usage，按「每次模型调用」累计， on_usage 回调吐出累计值（Web 控制台展示） """ import base64 import json import logging import httpx from fastmcp import Client from mcp_agent.config import AgentSettings _log = logging.getLogger("agent") S = AgentSettings() class _UsageUnsupported(RuntimeError): """模型/网关不认 stream_options.include_usage（400/422 或报错点名该字段）——降级重试用。""" SYSTEM_PROMPT = """你是手机自动化控制助手。你通过工具实时操作 Android 手机。 工作规范： 1. 先 de_list_devices 确定目标设备（在线才可操作）；设备有 name（名称）与 serial（地址）， **给用户汇报时用名称**（同型号多台靠它区分），调工具时仍用 serial 2. 观察屏幕：先 de_screenshot 获取截图（图像会随后给你），基于截图理解当前界面 3. 点击定位分优先级（不要自己推算像素坐标，那是精度最差的方式）： a) 目标有可见文字（按钮/菜单/列表标题/标签/输入框提示）→ de_tap_text 直接给文字， 一次完成「找到并点击」，原生控件与 WebView/图片渲染文字都支持 b) 文字有歧义或 de_tap_text 未命中 → de_ui_tree(limit=80) 看可点元素后 用 de_tap_element（text/text_contains 匹配） c) 只有纯图形目标（视频画面/无文字图标且树里没有）才用 de_tap 给坐标—— 坐标只需大致对准目标中心，服务端会自动吸附到该处可点击元素中心，无需精算 4. de_tap 点击后若返回 snapped=true 表示已吸附命中元素（可核对 label）； 截图判断界面变化=点击成功，无变化=未命中 5. 输入文字：先 de_tap_text 或 de_tap 点中输入框，再 de_type_text 输入 6. 每次关键操作后再次 de_screenshot 验证结果，直到完成用户目标 7. 若点击后截图无任何变化：不要重复点同一坐标，换 de_tap_text/de_tap_element 重新定位，或先 de_ui_tree 确认元素文案再试 8. 完成或失败时用中文总结：做了什么、当前状态、需要用户注意的事项 9. 设备不可用/操作失败时如实报告错误，不要臆测成功 10. 效率：界面未变化时不要重复截图/点击同一位置；每步都要推进目标； 若连续 6 步无进展（截图内容未变/操作无效），停止并总结原因，不要空转 可用工具清单将由系统提供。""" class Agent: def __init__(self, settings: AgentSettings = None): self.s = settings or S self.tools_schema = [] # OpenAI function schema self.messages = [] # 回调（Web 展示用，均可选）： # on_delta(text, kind) kind: content | reasoning —— 流式文本增量 # on_tool(step) step: {tool, args, result, image} —— 工具调用完成 # on_usage(usage) usage: {prompt_tokens, completion_tokens, # total_tokens, calls} —— 累计 token 用量 self.on_delta = None self.on_tool = None self.on_usage = None # 本轮累计用量（run_stream 开始时重置） self.usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0, "calls": 0} self._include_usage = True # 模型不认 stream_options 时自动置 False self._mcp = None # ---------- MCP 工具桥 ---------- async def _load_tools(self): """从 MCP Server 拉工具，转 OpenAI function schema。""" # 短连接超时：MCP 不可用时快速失败（默认会无限重试卡死线程） self._mcp = Client(self.s.mcp_url, timeout=10.0, init_timeout=10.0) await self._mcp.__aenter__() tools = await self._mcp.list_tools() self.tools_schema = [] for t in tools: # MCP SDK v2 改名 input_schema，兼容新旧字段 schema = getattr(t, "input_schema", None) or getattr(t, "inputSchema", {}) name = getattr(t, "name", "") desc = getattr(t, "description", "") or "" self.tools_schema.append({ "type": "function", "function": {"name": name, "description": desc, "parameters": schema}}) _log.info("MCP 工具已加载: %s", [s["function"]["name"] for s in self.tools_schema]) async def close(self): if self._mcp: try: await self._mcp.__aexit__(None, None, None) except Exception: pass # ---------- 模型调用（流式） ---------- async def _chat_stream(self): """流式 chat/completions：逐 chunk 产出 JSON（async generator）。 默认要求服务端在末尾 chunk 带 usage（token 统计）；个别网关不认 `stream_options` 会直接 400，此时自动降级重试一次（不影响主流程）。 """ if self._include_usage: try: async for chunk in self._chat_stream_once(True): yield chunk return except _UsageUnsupported as e: _log.warning("模型不支持 stream_options.include_usage，降级重试：%s", e) self._include_usage = False async for chunk in self._chat_stream_once(False): yield chunk async def _chat_stream_once(self, with_usage): body = { "model": self.s.model, "messages": self.messages, "tools": self.tools_schema if self.tools_schema else None, "max_tokens": 4096, "stream": True, } if with_usage: body["stream_options"] = {"include_usage": True} headers = {"Authorization": f"Bearer {self.s.api_key}", "Content-Type": "application/json"} url = f"{self.s.api_base.rstrip('/')}/chat/completions" async with httpx.AsyncClient(timeout=self.s.request_timeout) as client: async with client.stream("POST", url, json=body, headers=headers) as r: if r.status_code != 200: text = (await r.aread()).decode(errors="replace") # 本次开了 include_usage 却被打回（400/422，或报错里点名这个字段） # → 视作网关不支持，交给上层降级重试（401/余额等真错误照常抛出） if with_usage and (r.status_code in (400, 422) or "stream_options" in text): raise _UsageUnsupported(text[:200]) raise RuntimeError(f"模型 API HTTP {r.status_code}: {text[:300]}") async for line in r.aiter_lines(): if not line.startswith("data:"): continue data = line[5:].strip() if data == "[DONE]": break try: yield json.loads(data) except json.JSONDecodeError: continue # ---------- token 用量 ---------- @staticmethod def _read_usage(raw): """把一次模型调用返回的 usage 规整为 {prompt, completion, total}；无效返回 None。""" if not isinstance(raw, dict): return None try: pt = int(raw.get("prompt_tokens") or 0) ct = int(raw.get("completion_tokens") or 0) tt = int(raw.get("total_tokens") or 0) or (pt + ct) except (TypeError, ValueError): return None if not (pt or ct or tt): return None return {"prompt_tokens": pt, "completion_tokens": ct, "total_tokens": tt} def _accumulate_usage(self, raw): """把一次模型调用的 usage 累加进本轮总量，并回调 on_usage（累计值）。""" u = self._read_usage(raw) if not u: return self.usage["prompt_tokens"] += u["prompt_tokens"] self.usage["completion_tokens"] += u["completion_tokens"] self.usage["total_tokens"] += u["total_tokens"] self.usage["calls"] += 1 if self.on_usage: try: self.on_usage(dict(self.usage)) except Exception: pass # ---------- 工具执行 ---------- async def _execute_tool(self, name, arguments): """执行 MCP 工具，返回 (文本结果, image_data_or_None)。""" args = json.loads(arguments) if isinstance(arguments, str) else (arguments or {}) _log.info("执行工具 %s %s", name, args) try: result = await self._mcp.call_tool(name, args) data = getattr(result, "data", result) except Exception as e: return {"ok": False, "error": f"工具执行失败: {e}"}, None # de_screenshot：图像分离（作为 image_url 追加给模型看 + on_tool 缩略展示） image_b64 = None text_result = data if name == "de_screenshot" and isinstance(data, dict) and data.get("ok"): img = (data.get("data") or {}).get("image") or {} if img.get("data"): text_result = {k: v for k, v in (data.get("data") or {}).items() if k != "image"} image_b64 = img["data"] if self.on_tool: try: self.on_tool({"tool": name, "args": args, "result": text_result, "image": image_b64}) except Exception: pass return text_result, image_b64 def _repair_tool_messages(self): """修复 tool_calls 配对不完整：从尾部移除「assistant 带 tool_calls 但其后 tool 回应不足」的消息段（流中断可能丢失分片，400 重试前自愈）。""" for i in range(len(self.messages) - 1, -1, -1): m = self.messages[i] if m.get("role") == "assistant" and m.get("tool_calls"): # 统计其后 tool 消息数是否匹配 need = len(m["tool_calls"]) have = sum(1 for x in self.messages[i + 1:] if x.get("role") == "tool") if have < need: _log.warning("修复不完整 tool_calls 段（need=%d have=%d），回退 %d 条消息", need, have, len(self.messages) - i) self.messages = self.messages[:i] return # ---------- 主循环（流式） ---------- async def run_stream(self, prompt: str, serial: str = "", history=None, on_delta=None, on_tool=None, should_stop=None, extra_context=None, on_usage=None): """流式执行一轮指令，返回最终完整文本。 history：上一轮的 [{"role": "user"|"assistant", "content": 文本}] 列表， 用于多轮对话保持上下文（截图/工具消息不入历史，控制 token）。 on_delta(text, kind)：content/reasoning 文本增量（实时推给前端） on_tool(step)：工具调用完成（实时显示 MCP 步骤） should_stop：可调用 fn() -> bool，每轮模型调用前检查（用户中断用） extra_context：附加文本（经验记忆注入，放在 system prompt 末尾） on_usage(usage)：每完成一次模型调用回调一次（累计值，见 self.usage） 本轮累计 token 用量同时留在 self.usage（调用方可直接读）。 """ self.on_delta = on_delta self.on_tool = on_tool self.on_usage = on_usage self.usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0, "calls": 0} target = serial or self.s.default_serial sys_txt = SYSTEM_PROMPT if target: sys_txt += f"\n\n本次默认目标设备 serial：{target}（未指定设备时用它）。" if extra_context: sys_txt += f"\n\n## 过往成功经验参考（同类任务，可参考其中的操作套路，但要根据当前界面灵活调整）\n{extra_context}" self.messages = [{"role": "system", "content": sys_txt}] for h in (history or []): if h.get("role") in ("user", "assistant") and h.get("content"): self.messages.append({"role": h["role"], "content": h["content"]}) self.messages.append({"role": "user", "content": prompt}) for _step in range(self.s.max_steps): if should_stop and should_stop(): _log.info("Agent 被用户中断") return "（已按用户要求停止操作）" content_parts = [] tool_acc = {} # index -> {id, name, args} has_tool = False retried = False call_usage = None # 本次模型调用的 usage（末尾 chunk 带） while True: call_usage = None # 重试时丢弃上一次（未完成）的用量 try: async for chunk in self._chat_stream(): if chunk.get("usage"): call_usage = chunk["usage"] choice = (chunk.get("choices") or [{}])[0] delta = choice.get("delta") or {} text = delta.get("content") if text: content_parts.append(text) if on_delta: on_delta(text, "content") rtext = delta.get("reasoning_content") if rtext: if on_delta: on_delta(rtext, "reasoning") for tc in delta.get("tool_calls") or []: has_tool = True idx = tc.get("index", 0) acc = tool_acc.setdefault(idx, {"id": "", "name": "", "args": ""}) if tc.get("id"): acc["id"] = tc["id"] fn = tc.get("function") or {} if fn.get("name"): acc["name"] += fn["name"] if fn.get("arguments"): acc["args"] += fn["arguments"] break except RuntimeError as e: # 流中断导致 tool_calls 分片丢失：修复后重试一次 if ("tool_calls" in str(e) or "must be followed" in str(e)) and not retried: _log.warning("tool_calls 消息不完整，自愈重试") self._repair_tool_messages() retried = True continue raise self._accumulate_usage(call_usage) full_content = "".join(content_parts) if has_tool: # 组装 assistant 消息（含 tool_calls）并执行工具 tcs = [] for idx in sorted(tool_acc): acc = tool_acc[idx] tcs.append({"id": acc["id"] or f"call_{idx}", "type": "function", "function": {"name": acc["name"], "arguments": acc["args"]}}) self.messages.append({"role": "assistant", "content": full_content, "tool_calls": tcs}) for tc in tcs: fn = tc["function"] text_result, image_b64 = await self._execute_tool( fn["name"], fn["arguments"]) self.messages.append({ "role": "tool", "tool_call_id": tc["id"], "content": json.dumps(text_result, ensure_ascii=False)[:4000]}) if image_b64: self.messages.append({ "role": "user", "content": [{"type": "text", "text": "这是最新屏幕截图，请基于它继续判断"}, {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}}]}) continue # 无工具调用：本轮即最终回答 return full_content # 步骤超限：不带工具让模型做最终总结（避免机械提示，给用户有意义的结论） try: _log.warning("达到最大步骤数，请求模型收尾总结") saved_tools = self.tools_schema self.tools_schema = [] self.messages.append({"role": "user", "content": "已达最大操作步骤数，请立即用中文总结：" "已完成的部分、当前设备状态、未能完成的原因与下一步建议。" "不要调用任何工具。"}) parts = [] call_usage = None async for chunk in self._chat_stream(): if chunk.get("usage"): call_usage = chunk["usage"] delta = (chunk.get("choices") or [{}])[0].get("delta") or {} text = delta.get("content") if text: parts.append(text) if on_delta: on_delta(text, "content") self._accumulate_usage(call_usage) self.tools_schema = saved_tools summary = "".join(parts) return summary or "（已达步骤上限，模型未能生成总结）" except Exception as e: return f"（已达最大步骤数，且收尾总结失败: {e}）" # ---------- 非流式（CLI） ---------- async def run(self, prompt: str, serial: str = "") -> str: """非流式执行，返回最终文本（CLI 用，内部走流式收集）。""" return await self.run_stream(prompt, serial)