"""anuma.ai Responses SSE 事件 → IREvent（原生协议与 OpenAI Responses 一致）。
上游：``POST portal.anuma.ai/api/v1/responses``，返回标准 OpenAI Responses SSE：
``response.created`` / ``response.output_text.delta`` / ``response.completed`` 等。
正文增量由 ``response.output_text.delta`` 产出；思维链摘要增量由
``response.reasoning_summary_text.delta`` 产出（必须产出 kind="thinking"）。
**原生 function calling**：上游接受客户端 tools 并原生返回 function_call 事件
（``response.output_item.added`` type=function_call + ``response.function_call_arguments.delta/done``）。
本 parser 把 function_call 统一转成 ``{...}`` 围栏文本
（携带上游 call_id），走宿主既有解析链路（parse_tool_calls / ToolCallStreamParser）。
"""
from __future__ import annotations
import json
from typing import Any
from app.events import IREvent, Usage
from app.upstream.base import EventParser
class DefaultParser(EventParser):
def __init__(self) -> None:
# 流式 function_call 状态：call_id -> {name, arguments(字符串累积)}
self._pending_calls: dict[str, dict[str, str]] = {}
# 本轮是否已通过 output_text.delta 吐过正文（用于 done 去重）
self._saw_text_delta = False
def parse(self, raw: Any) -> list[IREvent]:
"""raw 为单个 SSE 事件 dict（``{"type": ..., ...}``），返回 0..n 个 IREvent。"""
if not isinstance(raw, dict):
return []
etype = raw.get("type") or ""
delta = raw.get("delta") or {}
if not isinstance(delta, dict):
delta = {}
events: list[IREvent] = []
# ---- 新响应开始：清本轮状态（parser 跨请求复用）----
if etype == "response.created":
self._pending_calls.clear()
self._saw_text_delta = False
return events
# ---- function_call 开始（output_item.added 里 type=function_call）----
if etype == "response.output_item.added":
item = raw.get("item") or {}
if isinstance(item, dict) and item.get("type") == "function_call":
call_id = str(item.get("call_id") or item.get("id") or "")
if call_id:
self._pending_calls[call_id] = {
"name": str(item.get("name") or ""),
"arguments": str(item.get("arguments") or ""),
}
return events
# ---- function_call arguments 增量 ----
if etype == "response.function_call_arguments.delta":
# delta 字段是 arguments 的字符串增量
for call_id, pending in self._pending_calls.items():
pending["arguments"] += str(delta if isinstance(delta, str) else delta.get("delta") or "")
return events
# ---- function_call arguments 完整 ----
if etype == "response.function_call_arguments.done":
for call_id, pending in self._pending_calls.items():
fence = _function_call_fence(pending["name"], pending["arguments"], call_id)
if fence:
events.append(IREvent(kind="text", text=fence))
self._pending_calls.clear()
return events
# ---- 思维链摘要增量（独立事件，kimi 等推理模型的思考链）----
# 2026-08-08 实测（HAR「聊天」kimi/kimi-k3）：kimi 的思考是**独立事件**
# response.reasoning_summary_text.delta（×128），结构：
# {"delta": {"OfString": "We", "OfResponseReasoningSummaryDeltaEventDelta": "We"},
# "type": "response.reasoning_summary_text.delta", ...}
# 两个字段都是**字符串**（GPT 系模型在 output_text.delta 里带的则是 dict）。
# 不解析 → 思考全丢 → 客户端等待期空白 + 正文瞬间喷出 = "一大段突然出现"。
if etype == "response.reasoning_summary_text.delta":
t = delta.get("OfString") if isinstance(delta, dict) else None
if not t:
s = delta.get("OfResponseReasoningSummaryDeltaEventDelta") if isinstance(delta, dict) else None
if isinstance(s, dict):
t = s.get("text") or ""
else:
t = s or ""
if t:
events.append(IREvent(kind="thinking", thinking=str(t)))
return events
# ---- 正文增量 ----
# 事件 type 是唯一语义来源：output_text.delta 只产出正文。
# 某些上游 union 序列化会把同一个正文 token 同时写入 OfString 和
# OfResponseReasoningSummaryDeltaEventDelta；后者无论字符串还是 dict 都不能在
# 此事件中再解释成 thinking，否则客户端会在正文后生成逐 token 的碎片思考卡。
if etype == "response.output_text.delta":
text = delta.get("OfString") if isinstance(delta, dict) else ""
if isinstance(text, dict): # 部分模型带 {"text": ...} 结构
text = text.get("text") or ""
if text:
self._saw_text_delta = True
events.append(IREvent(kind="text", text=str(text)))
return events
# ---- 输出完整文本（生图时图片 markdown 只在此事件的**顶层 text** 里）----
# 2026-08-08 实测（debug_parser.py）：生图响应里 /media/ 图片 URL（如
# ""）只
# 出现在 response.output_text.done 的**事件顶层 text 字段**（与 delta 同层，
# 不是 delta 里）。delta 事件只带思维链摘要，没有图片。不处理顶层 text
# 会把图片链接整个吞掉（此前生图只有 function_call 空壳、无媒体输出）。
#
# 聊天路径：delta 已流式吐完正文，done 再吐整段会重复（客户端先逐字再整段 dump）。
# 去重：已见 text delta 且 done 不含 media URL → 跳过；生图 / 无 delta 兜底照发。
if etype == "response.output_text.done":
text = raw.get("text")
if isinstance(text, str) and text:
if self._saw_text_delta and "portal.anuma.ai/api/v1/media" not in text:
return events
events.append(IREvent(kind="text", text=text))
return events
# ---- 完成事件：非流式 function_call 在 output 里；usage + finish ----
# 生图时图片签名 URL 在 response.tool_call_events[].output（JSON 字符串，
# 含 output_images[].url）——上游把生图结果当作工具执行结果回传。
# 2026-08-08 实测（debug_parser.py）：流式响应的 output_text 是空壳
# （{"type":"output_text"}），图片 markdown 只出现在 tool_call_events。
if etype == "response.completed":
resp = raw.get("response") or {}
if isinstance(resp, dict):
for item in resp.get("output") or []:
if isinstance(item, dict) and item.get("type") == "function_call":
fence = _function_call_fence(
str(item.get("name") or ""),
item.get("arguments") or "{}",
str(item.get("call_id") or item.get("id") or ""),
)
if fence:
events.append(IREvent(kind="text", text=fence))
# 生图结果：tool_call_events 里带 output 的调用 → 把图片 URL 以
# markdown 形式输出，客户端可见真实图片（不产出空壳调用围栏）。
for tce in resp.get("tool_call_events") or []:
if not isinstance(tce, dict):
continue
output = tce.get("output")
if not isinstance(output, str) or not output:
continue
try:
out = json.loads(output)
except json.JSONDecodeError:
continue
urls = []
for img in out.get("output_images") or []:
u = img.get("url") if isinstance(img, dict) else None
if u:
urls.append(u)
if urls:
text = "\n".join(f"" for u in urls)
events.append(IREvent(kind="text", text=text))
usage = self._parse_usage(raw)
if usage:
events.append(IREvent(kind="finish", usage_delta=usage, finish_reason="stop"))
else:
events.append(IREvent(kind="finish", finish_reason="stop"))
return events
# ---- 结尾独立 usage 事件（无 type 或 type 为 usage 的收尾事件）----
if etype == "" or etype == "response.usage":
usage = self._parse_usage(raw)
if usage:
events.append(IREvent(kind="finish", usage_delta=usage, finish_reason="stop"))
return events
# ---- 失败事件 ----
if etype == "response.failed":
err = raw.get("error") or raw.get("message") or "upstream response failed"
msg = err.get("message") if isinstance(err, dict) else str(err)
events.append(IREvent(kind="error", error=msg or "upstream response failed"))
return events
return events
@staticmethod
def _parse_usage(raw: dict[str, Any]) -> Usage | None:
usage = raw.get("usage")
if not isinstance(usage, dict):
return None
# anuma 返回 prompt_tokens/completion_tokens/total_tokens/cost_micro_usd/credits_used
# （OpenAI 标准为 input_tokens/output_tokens）
input_tokens = usage.get("input_tokens") or usage.get("prompt_tokens") or 0
output_tokens = usage.get("output_tokens") or usage.get("completion_tokens") or 0
if not input_tokens and not output_tokens:
return None
details = usage.get("output_tokens_details") or {}
thinking = details.get("reasoning_tokens") if isinstance(details, dict) else 0
cached = 0
in_details = usage.get("input_tokens_details") or {}
if isinstance(in_details, dict):
cached = in_details.get("cached_tokens") or 0
return Usage(
input_tokens=int(input_tokens),
output_tokens=int(output_tokens),
thinking_tokens=int(thinking or 0),
cached_tokens=int(cached or 0),
model=str(usage.get("model") or "") or None,
provider="anuma",
)
def _function_call_fence(name: str, arguments: str, call_id: str) -> str | None:
"""function_call → ``{"name","arguments","id"}`` 围栏文本。"""
if not name:
return None
try:
args = json.loads(arguments) if arguments.strip() else {}
except json.JSONDecodeError:
args = {"value": arguments}
if not isinstance(args, dict):
args = {"value": args}
return (
f"{json.dumps({'name': name, 'arguments': args, 'id': call_id},
ensure_ascii=False)}"
)