从 Responses API 请求中提取 messages 列表、tools 列表和 tool_choice。 返回: (messages, tools, tool_choice)
(data: dict)
| 147 | |
| 148 | |
| 149 | def extract_messages(data: dict): |
| 150 | """ |
| 151 | 从 Responses API 请求中提取 messages 列表、tools 列表和 tool_choice。 |
| 152 | 返回: (messages, tools, tool_choice) |
| 153 | """ |
| 154 | ROLE_MAP = {"developer": "system"} |
| 155 | raw_tools = data.get("tools", []) |
| 156 | tools = _convert_tools(raw_tools) |
| 157 | tool_choice = _convert_tool_choice(data.get("tool_choice")) |
| 158 | |
| 159 | if "input" not in data: |
| 160 | if "messages" in data: |
| 161 | return data["messages"], tools, tool_choice |
| 162 | return [], tools, tool_choice |
| 163 | |
| 164 | inp = data["input"] |
| 165 | if isinstance(inp, str): |
| 166 | messages = [] |
| 167 | if "instructions" in data and data["instructions"]: |
| 168 | messages.append({"role": "system", "content": data["instructions"]}) |
| 169 | messages.append({"role": "user", "content": inp}) |
| 170 | return messages, tools, tool_choice |
| 171 | |
| 172 | if not isinstance(inp, list): |
| 173 | return [], tools, tool_choice |
| 174 | |
| 175 | messages = [] |
| 176 | if "instructions" in data and data["instructions"]: |
| 177 | messages.append({"role": "system", "content": data["instructions"]}) |
| 178 | |
| 179 | pending_tool_calls = [] # 收集连续的 function_call |
| 180 | pending_reasoning = "" # function_call 携带的 reasoning_content |
| 181 | |
| 182 | def _flush_tool_calls(): |
| 183 | """将累积的 function_call 合并为一个 assistant 消息""" |
| 184 | nonlocal pending_tool_calls, pending_reasoning |
| 185 | if pending_tool_calls: |
| 186 | msg = { |
| 187 | "role": "assistant", |
| 188 | "content": "", |
| 189 | "tool_calls": pending_tool_calls, |
| 190 | } |
| 191 | if pending_reasoning: |
| 192 | msg["reasoning_content"] = pending_reasoning |
| 193 | messages.append(msg) |
| 194 | pending_tool_calls = [] |
| 195 | pending_reasoning = "" |
| 196 | |
| 197 | for item in inp: |
| 198 | if not isinstance(item, dict): |
| 199 | continue |
| 200 | item_type = item.get("type") |
| 201 | |
| 202 | if item_type == "message": |
| 203 | _flush_tool_calls() # 遇到非 function_call 项，先刷新累积的 tool calls |
| 204 | role = item.get("role", "user") |
| 205 | role = ROLE_MAP.get(role, role) |
| 206 | content = item.get("content", "") |
no test coverage detected