From 5458f3b8a40630844356e12e35ec6f421871852b Mon Sep 17 00:00:00 2001 From: showtan001 <240788545@qq.com> Date: Sun, 30 Aug 2026 22:29:19 +0800 Subject: [PATCH] Support Anthropic conversation logging and retries --- .../host_agent/conversation.py | 83 ++++++++++++++++++- apps/device-host-agent/host_agent/web/app.py | 19 ++++- runtime/ai_planner.py | 6 ++ runtime/task.py | 19 +++-- 4 files changed, 113 insertions(+), 14 deletions(-) diff --git a/apps/device-host-agent/host_agent/conversation.py b/apps/device-host-agent/host_agent/conversation.py index 64a3368..4e9cb7b 100644 --- a/apps/device-host-agent/host_agent/conversation.py +++ b/apps/device-host-agent/host_agent/conversation.py @@ -41,10 +41,10 @@ class ConversationAgent: self._client: Any | None = None def chat(self, messages: list[dict[str, Any]], *, tools: dict[str, Callable[..., Any]] | None = None) -> ChatResult: + if self.config.provider == "anthropic": + return self._chat_anthropic(messages, tools=tools) if self.config.provider not in {"openai", "openai_compatible"}: - raise RuntimeError( - "conversation chat currently requires an OpenAI-compatible provider" - ) + raise RuntimeError("unsupported conversation provider") client = self._get_client() active_tools = tools or self.tools history: list[dict[str, Any]] = [ @@ -104,11 +104,55 @@ class ConversationAgent: ) raise RuntimeError("conversation exceeded the maximum tool-call rounds") + def _chat_anthropic(self, messages: list[dict[str, Any]], *, tools: dict[str, Callable[..., Any]] | None = None) -> ChatResult: + import anthropic + + kwargs: dict[str, Any] = {} + if self.config.api_key: + kwargs["api_key"] = self.config.api_key + if self.config.base_url: + kwargs["base_url"] = self.config.base_url + client = anthropic.Anthropic(**kwargs) + history = [_normalize_anthropic_message(message) for message in messages] + active_tools = tools or self.tools + calls = 0 + for _ in range(self.max_rounds): + response = client.messages.create( + model=self.config.resolved_model(), + max_tokens=2048, + system="You are a mobile device assistant. Reply naturally, or call one device tool when needed. Never claim an action succeeded unless its tool result confirms it.", + messages=history, + tools=[_anthropic_tool(spec) for spec in [*ACTION_TOOL_SPECS, *READ_TOOL_SPECS]], + timeout=self.config.timeout, + ) + blocks = getattr(response, "content", []) or [] + text_parts = [getattr(block, "text", "") for block in blocks if getattr(block, "type", None) == "text"] + thinking_parts = [getattr(block, "thinking", "") for block in blocks if getattr(block, "type", None) == "thinking"] + uses = [block for block in blocks if getattr(block, "type", None) == "tool_use"] + self._log({"type": "llm_response", "content": "\n".join(x for x in text_parts if x), "thinking": "\n".join(x for x in thinking_parts if x), "tool_calls": [{"id": u.id, "name": u.name, "arguments": u.input} for u in uses]}) + if not uses: + content = "\n".join(x for x in text_parts if x) + self._log({"type": "final_reply", "content": content}) + return ChatResult(content=content, tool_calls=calls) + history.append({"role": "assistant", "content": [_anthropic_block_dict(block) for block in blocks if getattr(block, "type", None) != "thinking"]}) + results = [] + for use in uses: + arguments = dict(use.input) if isinstance(use.input, dict) else {} + try: + result = active_tools[use.name](**arguments) + except Exception as exc: + result = {"ok": False, "error": str(exc)} + calls += 1 + self._log({"type": "tool_result", "tool_call_id": use.id, "tool_name": use.name, "arguments": arguments, "result": result}) + results.append({"type": "tool_result", "tool_use_id": use.id, "content": json.dumps(result, ensure_ascii=True, default=str)}) + history.append({"role": "user", "content": results}) + raise RuntimeError("conversation exceeded the maximum tool-call rounds") + def chat_for_device(self, device_id: str, messages: list[dict[str, Any]]) -> ChatResult: """Run a conversation bound to exactly one device.""" device_tools = { name: _bind_device(tool, device_id, name=name) - for name, tool in original_tools.items() + for name, tool in self.tools.items() } return self.chat(messages, tools=device_tools) @@ -150,6 +194,37 @@ def _openai_tool(spec: ToolSpec) -> dict[str, Any]: } +def _anthropic_tool(spec: ToolSpec) -> dict[str, Any]: + return {"name": spec.name, "description": spec.description, "input_schema": spec.parameters} + + +def _anthropic_block_dict(block: Any) -> dict[str, Any]: + block_type = getattr(block, "type", "") + if block_type == "text": + return {"type": "text", "text": getattr(block, "text", "")} + if block_type == "thinking": + return {"type": "thinking", "thinking": getattr(block, "thinking", "")} + return {"type": "tool_use", "id": block.id, "name": block.name, "input": block.input} + + +def _normalize_anthropic_message(message: dict[str, Any]) -> dict[str, Any]: + normalized = _normalize_message(message) + content = normalized["content"] + if isinstance(content, str): + return normalized + blocks = [] + for block in content: + if block.get("type") == "text": + blocks.append(block) + elif block.get("type") == "image_url": + url = block.get("image_url", {}).get("url", "") + if isinstance(url, str) and url.startswith("data:"): + header, data = url.split(",", 1) + media_type = header[5:].split(";", 1)[0] + blocks.append({"type": "image", "source": {"type": "base64", "media_type": media_type, "data": data}}) + return {"role": normalized["role"], "content": blocks} + + def _message_dict(message: Any) -> dict[str, Any]: result: dict[str, Any] = {"role": "assistant"} content = getattr(message, "content", None) diff --git a/apps/device-host-agent/host_agent/web/app.py b/apps/device-host-agent/host_agent/web/app.py index 8d3ae7d..9c99baa 100644 --- a/apps/device-host-agent/host_agent/web/app.py +++ b/apps/device-host-agent/host_agent/web/app.py @@ -444,9 +444,22 @@ def create_console_app( ] if not messages: raise HTTPException(status_code=400, detail="messages are invalid") - result = await asyncio.to_thread( - conversation_agent.chat_for_device, device_id.strip(), messages - ) + if conversation_log is not None: + await asyncio.to_thread( + conversation_log.append, + {"type": "user_request", "device_id": device_id.strip(), "messages": messages}, + ) + try: + result = await asyncio.to_thread( + conversation_agent.chat_for_device, device_id.strip(), messages + ) + except Exception as exc: + if conversation_log is not None: + await asyncio.to_thread( + conversation_log.append, + {"type": "agent_error", "device_id": device_id.strip(), "error": str(exc)}, + ) + raise HTTPException(status_code=502, detail=str(exc)) from exc return JSONResponse( {"content": result.content, "tool_calls": result.tool_calls} ) diff --git a/runtime/ai_planner.py b/runtime/ai_planner.py index 09d8706..961300d 100644 --- a/runtime/ai_planner.py +++ b/runtime/ai_planner.py @@ -43,6 +43,12 @@ class AIPlanner(Planner): history_summary=_history_summary(world), device_platform=context.device_platform, ) + if context.step_results and not context.step_results[-1].success: + user_prompt += ( + "\n\nThe previous tool call failed. Diagnose the failure and choose a " + "corrected action or finish the task if it cannot proceed.\n" + f"Previous failure: {context.step_results[-1].error or 'unknown error'}" + ) decision = self.client.decide( system_prompt=PLANNER_SYSTEM_PROMPT, user_prompt=user_prompt, diff --git a/runtime/task.py b/runtime/task.py index 398ddee..631994a 100644 --- a/runtime/task.py +++ b/runtime/task.py @@ -187,13 +187,18 @@ class TaskRunner: "failed", result.error or "step failed", ) - self._update_task( - task, - status="failed", - completed=True, - failure_reason=result.error or "step failed", - ) - return task + # AI planners can use the structured failure feedback to + # correct malformed arguments or choose another action. + # Deterministic planners retain their fail-fast behavior. + if self.planner.__class__.__name__ != "AIPlanner": + self._update_task( + task, + status="failed", + completed=True, + failure_reason=result.error or "step failed", + ) + return task + break self._emit_step_progress( len(context.step_results),