from __future__ import annotations import json from dataclasses import dataclass, field from typing import Any, Dict, List, Optional from core.environment.context_builder import EnvironmentInjection @dataclass class ResponsesRuntimeRequest: user_input: str model: str instructions: Optional[str] = None environment: Optional[EnvironmentInjection] = None tools: List[Dict[str, Any]] = field(default_factory=list) input_items: Optional[List[Dict[str, Any]]] = None previous_response_id: Optional[str] = None reasoning_effort: Optional[str] = None max_output_tokens: Optional[int] = None text_format: Optional[Dict[str, Any]] = None tool_choice: Any = "auto" class ResponsesAgentRuntime: """ 新版主 Agent runtime 骨架。 目前先負責： 1. 統一組裝 Responses API payload 2. 固定附帶 environment injection 3. 為 hosted tools / bridge tools 預留同一個組裝入口 """ def build_request_payload(self, request: ResponsesRuntimeRequest) -> Dict[str, Any]: input_parts: List[Dict[str, Any]] = list(request.input_items or []) if request.environment: input_parts.insert( 0, { "role": "system", "content": [ { "type": "input_text", "text": "Latest environment context:\n" + request.environment.summary_text, } ], } ) input_parts.append( self.message_to_input_item({"role": "user", "content": request.user_input}) ) payload: Dict[str, Any] = { "model": request.model, "input": input_parts, "tools": self.normalize_tools_for_responses(request.tools), } if request.instructions: payload["instructions"] = request.instructions if request.previous_response_id: payload["previous_response_id"] = request.previous_response_id if request.reasoning_effort: payload["reasoning"] = {"effort": request.reasoning_effort} if request.max_output_tokens: payload["max_output_tokens"] = request.max_output_tokens if request.text_format: payload["text"] = {"format": request.text_format} if request.tools: payload["tool_choice"] = request.tool_choice return payload def build_payload_from_messages( self, *, messages: List[Dict[str, Any]], model: str, tools: Optional[List[Dict[str, Any]]] = None, reasoning_effort: Optional[str] = None, max_output_tokens: Optional[int] = None, tool_choice: Any = "auto", text_format: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: input_items: List[Dict[str, Any]] = [] instructions: Optional[str] = None for message in messages: if message.get("role") == "system": content = message.get("content") or "" instructions = f"{instructions}\n\n{content}" if instructions else str(content) continue input_items.append(self.message_to_input_item(message)) payload: Dict[str, Any] = { "model": model, "input": input_items, "tools": self.normalize_tools_for_responses(tools or []), } if instructions: payload["instructions"] = instructions if reasoning_effort: payload["reasoning"] = {"effort": reasoning_effort} if max_output_tokens: payload["max_output_tokens"] = max_output_tokens if text_format: payload["text"] = {"format": text_format} if tools: payload["tool_choice"] = tool_choice return payload @staticmethod def without_hosted_tools(payload: Dict[str, Any]) -> Dict[str, Any]: stripped = dict(payload) tools = [ tool for tool in stripped.get("tools", []) if tool.get("type") == "function" ] stripped["tools"] = tools if not tools: stripped.pop("tool_choice", None) return stripped @staticmethod def message_to_input_item(message: Dict[str, Any]) -> Dict[str, Any]: role = message.get("role") or "user" content = message.get("content") or "" if isinstance(content, list): return {"role": role, "content": [ResponsesAgentRuntime.normalize_content_part(part, role) for part in content]} content_type = "output_text" if role == "assistant" else "input_text" return {"role": role, "content": [{"type": content_type, "text": str(content)}]} @staticmethod def normalize_content_part(part: Dict[str, Any], role: str) -> Dict[str, Any]: part_type = part.get("type") if part_type == "text": return { "type": "output_text" if role == "assistant" else "input_text", "text": str(part.get("text", "")), } if part_type == "image_url": image_url = part.get("image_url") or {} return { "type": "input_image", "image_url": image_url.get("url", image_url if isinstance(image_url, str) else ""), } return dict(part) @staticmethod def normalize_tools_for_responses(tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]: normalized: List[Dict[str, Any]] = [] for tool in tools: if tool.get("type") != "function" or "function" not in tool: normalized.append(dict(tool)) continue fn = tool.get("function") or {} converted = { "type": "function", "name": fn.get("name"), "description": fn.get("description", ""), "parameters": fn.get("parameters", {"type": "object", "properties": {}}), } if "strict" in fn: converted["strict"] = fn["strict"] normalized.append(converted) return normalized @staticmethod def extract_output_text(response: Any) -> str: text = getattr(response, "output_text", None) if isinstance(text, str) and text.strip(): return text.strip() parts: List[str] = [] for item in getattr(response, "output", []) or []: item_type = getattr(item, "type", None) if item_type != "message": continue for content in getattr(item, "content", []) or []: content_text = getattr(content, "text", None) if content_text: parts.append(str(content_text)) return "\n".join(parts).strip() @staticmethod def extract_function_calls(response: Any) -> List[Dict[str, Any]]: calls: List[Dict[str, Any]] = [] for item in getattr(response, "output", []) or []: if getattr(item, "type", None) != "function_call": continue calls.append( { "id": getattr(item, "call_id", None) or getattr(item, "id", None), "type": "function", "function": { "name": getattr(item, "name", ""), "arguments": getattr(item, "arguments", "{}") or "{}", }, } ) return calls @staticmethod def decode_arguments(arguments: str) -> Dict[str, Any]: try: return json.loads(arguments or "{}") except json.JSONDecodeError: return {}