qzl/eryao
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: 更新协议文档，删除废弃计划文档

Browse Source 
- 更新 http-error-codes, user-points-chat-data-protocol
- 更新 divination-run-protocol, profile-protocol
- 删除废弃的后端和前端设计计划文档
This commit is contained in:
qzl
2026-04-08 17:23:02 +08:00
parent 49fc9a116f
commit e80a82bef4
57 changed files with 4117 additions and 2269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/core/agentscope/events/agui_codec.py
+1 -15
View File
@@ -11,8 +11,6 @@ from ag_ui.core import (
StepStartedEvent,
StepFinishedEvent,
)
from core.agentscope.runtime.ui_compiler import compile as compile_ui_hints
from schemas.agent.ui_hints import UiHintsPayload
if TYPE_CHECKING:
pass
@@ -55,15 +53,6 @@ def _sanitize_agui_event(event: dict[str, Any]) -> dict[str, Any]:
payload = dict(event)
event_type = str(payload.get("type", "")).strip().upper()
if event_type == EventType.TEXT_MESSAGE_END.value:
ui_hints = payload.get("ui_hints")
if ui_hints is not None:
try:
ui_hints_payload = UiHintsPayload.model_validate(ui_hints)
ui_schema = compile_ui_hints(ui_hints_payload)
payload["ui_schema"] = ui_schema
except Exception:
pass
payload.pop("ui_hints", None)
for key in (
"inputTokens",
"outputTokens",
@@ -72,9 +61,6 @@ def _sanitize_agui_event(event: dict[str, Any]) -> dict[str, Any]:
"model",
):
payload.pop(key, None)
if event_type == EventType.TOOL_CALL_RESULT.value:
payload.pop("ui_hints", None)
payload.pop("ui_schema", None)
return payload
@@ -182,7 +168,7 @@ def to_agui_wire_event(event: dict[str, Any] | BaseEvent) -> dict[str, Any]:
tool_result_payload["threadId"] = thread_id
if isinstance(run_id, str) and run_id:
tool_result_payload["runId"] = run_id
reserved = {"type", "threadId", "runId", "ui_hints", "ui_schema"}
reserved = {"type", "threadId", "runId"}
tool_result_payload.update({k: v for k, v in data.items() if k not in reserved})
return tool_result_payload
backend/src/core/agentscope/events/store.py
+34 -43
View File
@@ -13,7 +13,7 @@ from core.logging import get_logger
from schemas.agent.forwarded_props import RuntimeMode
from schemas.enums import AgentChatMessageRole, AgentChatSessionStatus
from schemas.agent.system_agent import AgentType
from schemas.agent.runtime_models import AgentOutput, ToolAgentOutput
from schemas.agent.runtime_models import AgentOutput, FollowUpOutput, ToolAgentOutput
from schemas.agent.visibility import SystemVisibilityBit, bit_mask
from schemas.domain.chat_message import AgentChatMessageMetadata
@@ -103,8 +103,6 @@ class SqlAlchemyEventStore:
session_repo: SessionRepository,
message_repo: MessageRepository,
) -> None:
message_id_raw = self._event_value(event, "messageId")
message_id = message_id_raw if isinstance(message_id_raw, str) else ""
content_value = self._event_value(event, "answer")
content = content_value if isinstance(content_value, str) else ""
if not content:
@@ -122,17 +120,22 @@ class SqlAlchemyEventStore:
if run_id_value is None:
return
runtime_mode = self._resolve_runtime_mode(event=event)
worker_output_fields = (
"status",
"sign_level",
"conclusion",
"focus_points",
"advice",
"keywords",
"answer",
"error",
"divination_derived",
"ui_hints",
(
"status",
"sign_level",
"conclusion",
"focus_points",
"advice",
"keywords",
"answer",
"error",
"divination_derived",
)
if runtime_mode == RuntimeMode.CHAT.value
else ("status", "answer", "error")
)
worker_output_payload: dict[str, object] = {}
for field in worker_output_fields:
@@ -143,21 +146,16 @@ class SqlAlchemyEventStore:
if not worker_output_payload:
return
try:
if runtime_mode == RuntimeMode.CHAT.value:
worker_output = AgentOutput.model_validate(worker_output_payload)
agent_type = AgentType.WORKER
metadata_model = AgentChatMessageMetadata(
run_id=run_id_value,
agent_type=agent_type,
agent_output=worker_output,
)
except Exception:
self._logger.warning(
"invalid worker metadata payload",
run_id=run_id_value,
message_id=message_id,
)
return
else:
worker_output = FollowUpOutput.model_validate(worker_output_payload)
agent_type = AgentType.WORKER
metadata_model = AgentChatMessageMetadata(
run_id=run_id_value,
agent_type=agent_type,
agent_output=worker_output,
)
role_value = self._event_value(event, "role")
if not isinstance(role_value, str):
@@ -345,29 +343,22 @@ class SqlAlchemyEventStore:
if isinstance(metadata, dict):
message_payload["metadata"] = metadata
try:
context_cache = create_context_messages_cache()
await context_cache.append_message(
thread_id=str(session_id),
runtime_mode=self._resolve_runtime_mode(event=event),
visibility_mask=visibility_mask,
message=message_payload,
)
except Exception as exc:
self._logger.warning(
"Failed to append context cache message from event",
thread_id=str(session_id),
error=str(exc),
)
context_cache = create_context_messages_cache()
await context_cache.append_message(
thread_id=str(session_id),
runtime_mode=self._resolve_runtime_mode(event=event),
visibility_mask=visibility_mask,
message=message_payload,
)
@staticmethod
def _resolve_runtime_mode(*, event: dict[str, Any]) -> str:
raw = event.get("runtime_mode")
if isinstance(raw, str):
normalized = raw.strip().lower()
if normalized:
if normalized in (RuntimeMode.CHAT.value, RuntimeMode.FOLLOW_UP.value):
return normalized
return RuntimeMode.CHAT.value
raise ValueError("invalid runtime_mode in event payload")
@staticmethod
def _resolve_message_timestamp(message: Any) -> str:
backend/src/core/agentscope/runtime/runner.py
+52 -34
View File
@@ -35,6 +35,7 @@ from schemas.agent.forwarded_props import (
)
from schemas.domain.divination import DerivedDivinationData
from schemas.agent.runtime_models import (
FollowUpOutput,
WorkerAgentOutputLite,
resolve_worker_output_model,
)
@@ -102,21 +103,23 @@ class AgentScopeRunner:
worker_toolkit = self._build_toolkit()
if cancel_checker is not None and await cancel_checker():
raise asyncio.CancelledError("run canceled by user")
derived_divination = self._resolve_derived_divination(
run_input=run_input
)
await self._emit_step_event(
pipeline=pipeline,
run_input=run_input,
step_name="divination",
event_type="DIVINATION_DERIVED",
runtime_mode=runtime_mode,
extra_event={
"divination": derived_divination.model_dump(
mode="json", by_alias=True, exclude_none=True
)
},
)
derived_divination: DerivedDivinationData | None = None
if runtime_mode == RuntimeMode.CHAT:
derived_divination = self._resolve_derived_divination(
run_input=run_input
)
await self._emit_step_event(
pipeline=pipeline,
run_input=run_input,
step_name="divination",
event_type="DIVINATION_DERIVED",
runtime_mode=runtime_mode,
extra_event={
"divination": derived_divination.model_dump(
mode="json", by_alias=True, exclude_none=True
)
},
)
worker_output = await self._execute_worker_step(
pipeline=pipeline,
run_input=run_input,
@@ -208,9 +211,11 @@ class AgentScopeRunner:
stage_config: SystemAgentRuntimeConfig,
runtime_client_time: ClientTimeContext | None,
runtime_mode: RuntimeMode,
derived_divination: DerivedDivinationData,
) -> WorkerAgentOutputLite:
worker_output_model = resolve_worker_output_model()
derived_divination: DerivedDivinationData | None,
) -> WorkerAgentOutputLite | FollowUpOutput:
worker_output_model = resolve_worker_output_model(
runtime_mode=runtime_mode.value
)
await self._emit_step_event(
pipeline=pipeline,
run_input=run_input,
@@ -252,11 +257,11 @@ class AgentScopeRunner:
toolkit: Any,
run_input: RunAgentInput,
stage_config: SystemAgentRuntimeConfig,
worker_output_model: type[WorkerAgentOutputLite],
worker_output_model: type[WorkerAgentOutputLite | FollowUpOutput],
pipeline: PipelineLike,
runtime_client_time: ClientTimeContext | None,
runtime_mode: RuntimeMode,
derived_divination: DerivedDivinationData,
derived_divination: DerivedDivinationData | None,
) -> StageExecutionResult:
tracking_model = self._build_model(stage_config=stage_config)
formatter = OpenAIChatFormatter()
@@ -292,12 +297,11 @@ class AgentScopeRunner:
usage_summary=tracking_model.usage_summary(),
)
await emitter.emit_final_text_end(
worker_output={
**worker_payload.model_dump(mode="json", exclude_none=True),
"divination_derived": derived_divination.model_dump(
mode="json", by_alias=True, exclude_none=True
),
},
worker_output=self._build_final_worker_output(
worker_payload=worker_payload,
runtime_mode=runtime_mode,
derived_divination=derived_divination,
),
response_metadata=response_metadata,
)
return StageExecutionResult(
@@ -316,15 +320,18 @@ class AgentScopeRunner:
*,
context_messages: list[Msg],
run_input: RunAgentInput,
derived_divination: DerivedDivinationData,
derived_divination: DerivedDivinationData | None,
) -> list[Msg]:
if context_messages:
last = context_messages[-1]
if last.role == "user":
return context_messages
_, _ = extract_latest_user_payload(run_input)
user_text = build_divination_user_prompt(derived=derived_divination)
_, latest_user_text = extract_latest_user_payload(run_input)
if derived_divination is None:
user_text = latest_user_text
else:
user_text = build_divination_user_prompt(derived=derived_divination)
user_blocks = [{"type": "text", "text": user_text}]
if (
user_blocks
@@ -422,12 +429,23 @@ class AgentScopeRunner:
@staticmethod
def _resolve_runtime_mode(*, run_input: RunAgentInput) -> RuntimeMode:
try:
return parse_forwarded_props_runtime_mode(
getattr(run_input, "forwarded_props", None)
return parse_forwarded_props_runtime_mode(
getattr(run_input, "forwarded_props", None)
)
@staticmethod
def _build_final_worker_output(
*,
worker_payload: WorkerAgentOutputLite | FollowUpOutput,
runtime_mode: RuntimeMode,
derived_divination: DerivedDivinationData | None,
) -> dict[str, Any]:
payload = worker_payload.model_dump(mode="json", exclude_none=True)
if runtime_mode == RuntimeMode.CHAT and derived_divination is not None:
payload["divination_derived"] = derived_divination.model_dump(
mode="json", by_alias=True, exclude_none=True
)
except ValueError:
return RuntimeMode.CHAT
return payload
@staticmethod
def _resolve_provider_api_key(*, factory_name: str) -> str:
backend/src/core/agentscope/runtime/stage_emitter.py
+11 -9
View File
@@ -57,19 +57,21 @@ class PipelineStageEmitter:
"role": "assistant",
"stage": self._stage,
"status": worker_output.get("status"),
"sign_level": worker_output.get("sign_level"),
"conclusion": worker_output.get("conclusion", []),
"focus_points": worker_output.get("focus_points", []),
"advice": worker_output.get("advice", []),
"keywords": worker_output.get("keywords", []),
"answer": worker_output.get("answer", ""),
"error": worker_output.get("error"),
"divination_derived": worker_output.get("divination_derived"),
**response_metadata,
}
ui_hints = worker_output.get("ui_hints")
if ui_hints is not None:
payload["ui_hints"] = ui_hints
if self._runtime_mode == "chat":
payload.update(
{
"sign_level": worker_output.get("sign_level"),
"conclusion": worker_output.get("conclusion", []),
"focus_points": worker_output.get("focus_points", []),
"advice": worker_output.get("advice", []),
"keywords": worker_output.get("keywords", []),
"divination_derived": worker_output.get("divination_derived"),
}
)
await self._emit("TEXT_MESSAGE_END", payload)
async def _emit_text_events_from_msg(self, msg: Msg) -> None:
backend/src/core/agentscope/runtime/tasks.py
+78 -7
View File
@@ -6,6 +6,7 @@ from typing import Any, cast
from uuid import UUID
from agentscope.message import Msg
from pydantic import TypeAdapter
from core.agentscope.caches import create_user_context_cache
from core.agentscope.caches.attachment_content_cache import (
create_attachment_content_cache,
@@ -31,6 +32,7 @@ from schemas.agent.forwarded_props import (
parse_forwarded_props_runtime_mode,
)
from schemas.agent.runtime_config import MessageContextConfig, RuntimeConfig
from schemas.agent.runtime_models import RuntimeAgentOutput
from schemas.domain.chat_message import (
AgentChatMessageMetadata,
extract_user_message_attachments,
@@ -46,6 +48,7 @@ from v1.points.service import PointsService
logger = get_logger("core.agentscope.runtime.tasks")
_MAX_CONTEXT_ATTACHMENTS = 3
_RUNTIME_AGENT_OUTPUT_ADAPTER = TypeAdapter(RuntimeAgentOutput)
def _serialize_tool_agent_output(
@@ -75,6 +78,72 @@ def _serialize_tool_agent_output(
)
def _serialize_assistant_context_from_metadata(
*,
metadata: AgentChatMessageMetadata | dict[str, object] | None,
fallback_content: str,
) -> str:
if metadata is None:
return fallback_content
try:
resolved_metadata = (
metadata
if isinstance(metadata, AgentChatMessageMetadata)
else AgentChatMessageMetadata.model_validate(metadata)
)
except Exception:
return fallback_content
agent_output = resolved_metadata.agent_output
if agent_output is None and isinstance(metadata, dict):
raw = metadata.get("agent_output")
if raw is not None:
try:
agent_output = _RUNTIME_AGENT_OUTPUT_ADAPTER.validate_python(raw)
except Exception:
return fallback_content
if agent_output is None:
return fallback_content
payload = agent_output.model_dump(mode="json", by_alias=True, exclude_none=True)
lines: list[str] = ["[assistant_context]"]
answer = payload.get("answer")
if isinstance(answer, str) and answer:
lines.append(f"answer: {answer}")
sign_level = payload.get("sign_level")
if isinstance(sign_level, str) and sign_level:
lines.append(f"sign_level: {sign_level}")
focus_points = payload.get("focus_points")
if isinstance(focus_points, list) and focus_points:
lines.append("focus_points:")
for item in focus_points:
if isinstance(item, str) and item:
lines.append(f"- {item}")
divination_derived = payload.get("divination_derived")
if isinstance(divination_derived, dict) and divination_derived:
lines.append("divination_derived:")
key_map = (
("guaName", "gua_name"),
("targetGuaName", "target_gua_name"),
("binaryCode", "binary_code"),
("changedBinaryCode", "changed_binary_code"),
)
for key, label in key_map:
value = divination_derived.get(key)
if isinstance(value, str) and value:
lines.append(f" {label}: {value}")
if len(lines) <= 1:
return fallback_content
return "\n".join(lines)
def _load_runtime() -> type[Any]:
return AgentScopeRuntimeOrchestrator
@@ -235,6 +304,12 @@ async def _build_recent_context_messages(
continue
content = tool_content
if role == "assistant":
content = _serialize_assistant_context_from_metadata(
metadata=metadata,
fallback_content=content,
)
converted.append(
Msg(
name=role or "user",
@@ -259,13 +334,9 @@ async def run_agentscope_task(command: dict[str, Any]) -> dict[str, object]:
raise ValueError("run_input is required")
run_input = parse_run_input(run_input_raw)
runtime_mode = RuntimeMode.CHAT
try:
runtime_mode = parse_forwarded_props_runtime_mode(
getattr(run_input, "forwarded_props", None)
)
except ValueError:
runtime_mode = RuntimeMode.CHAT
runtime_mode = parse_forwarded_props_runtime_mode(
getattr(run_input, "forwarded_props", None)
)
runtime_config = RuntimeConfig.model_validate(runtime_config_raw or {})
thread_id = run_input.thread_id
run_id = run_input.run_id
backend/src/core/agentscope/runtime/ui_compiler.py
-638
View File
@@ -1,638 +0,0 @@
"""
UiCompiler - 将描述性 UiHints 编译为渲染性 UiSchemaRenderer
设计原则:
- 机械转换: 不依赖复杂语义理解
- 尽量无损: hints 中出现的主要内容字段尽量保留
- 弱模板: intent 只影响默认布局，不决定字段是否丢失
"""
from __future__ import annotations
from typing import Any
from schemas.agent.ui_hints import (
UiHintAction,
UiHintActionTarget,
UiHintIcon,
UiHintIntent,
UiHintKvItem,
UiHintListItem,
UiHintSection,
UiHintsPayload,
UiHintStatus,
UiHintTextFormat,
)
# ============================================================
# Helpers
# ============================================================
def _compact(value: Any) -> Any:
"""递归移除 dict/list 中的 None，保留 False/0/空字符串/空列表按原样存在。"""
if isinstance(value, dict):
return {k: _compact(v) for k, v in value.items() if v is not None}
if isinstance(value, list):
return [_compact(v) for v in value]
return value
def _non_empty(nodes: list[dict[str, Any] | None]) -> list[dict[str, Any]]:
return [node for node in nodes if node is not None]
def compile_status(status: UiHintStatus) -> str:
return status.value
def _status_badge_needed(intent: UiHintIntent, status: UiHintStatus) -> bool:
return intent == UiHintIntent.STATUS or status != UiHintStatus.INFO
def _status_label(status: str) -> str:
normalized = status.strip().lower()
if not normalized:
return "ui.status.info"
return f"ui.status.{normalized}"
# ============================================================
# Action Compilation
# ============================================================
def compile_action(action: UiHintActionTarget) -> dict[str, Any]:
"""
编译 action target。
关键修复:
- 使用 by_alias=True，避免 toolId/successMessage/submitTo 丢失
"""
action_dict = action.model_dump(by_alias=True, exclude_none=True)
action_type = action_dict["type"]
if action_type == "navigation":
result: dict[str, Any] = {
"type": "navigation",
"path": action_dict["path"],
}
if "params" in action_dict:
result["params"] = action_dict["params"]
return result
if action_type == "url":
return {
"type": "url",
"url": action_dict["url"],
"target": action_dict.get("target", "_blank"),
}
if action_type == "event":
result = {
"type": "event",
"event": action_dict["event"],
}
if "payload" in action_dict:
result["payload"] = action_dict["payload"]
return result
if action_type == "tool":
result = {
"type": "tool",
"toolId": action_dict["toolId"],
}
if "params" in action_dict:
result["params"] = action_dict["params"]
return result
if action_type == "copy":
result = {
"type": "copy",
"content": action_dict["content"],
}
if "successMessage" in action_dict:
result["successMessage"] = action_dict["successMessage"]
return result
if action_type == "payload":
result = {
"type": "payload",
"payload": action_dict["payload"],
}
if "submitTo" in action_dict:
result["submitTo"] = action_dict["submitTo"]
return result
raise ValueError(f"Unknown action type: {action_type}")
def compile_button(action: UiHintAction) -> dict[str, Any]:
return _compact(
{
"type": "button",
"label": action.label,
"style": action.style.value if action.style else "primary",
"disabled": action.disabled,
"action": compile_action(action.action),
}
)
# ============================================================
# Small Node Compilation
# ============================================================
def compile_icon_spec(icon: UiHintIcon) -> dict[str, Any]:
return _compact(
{
"source": icon.source.value,
"value": icon.value,
"color": icon.color,
"size": icon.size,
}
)
def compile_icon_node(icon: UiHintIcon) -> dict[str, Any]:
return _compact(
{
"type": "icon",
"source": icon.source.value,
"value": icon.value,
"color": icon.color,
"size": icon.size,
}
)
def compile_text(
content: str,
*,
role: str = "body",
format: UiHintTextFormat | str = UiHintTextFormat.PLAIN,
status: str | None = None,
) -> dict[str, Any]:
fmt = format.value if isinstance(format, UiHintTextFormat) else format
return _compact(
{
"type": "text",
"content": content,
"format": fmt,
"role": role,
"status": status,
}
)
def compile_badge(label: str, status: str) -> dict[str, Any]:
return {
"type": "badge",
"label": label,
"status": status,
}
def compile_kv_item(item: UiHintKvItem) -> dict[str, Any]:
return _compact(
{
"key": item.key,
"label": item.label,
"value": item.value,
"copyable": item.copyable,
}
)
def compile_kv(items: list[UiHintKvItem], columns: int = 1) -> dict[str, Any]:
return {
"type": "kv",
"items": [compile_kv_item(i) for i in items],
"columns": columns,
}
def compile_divider() -> dict[str, Any]:
return {"type": "divider", "inset": 0}
# ============================================================
# Layout Compilation
# ============================================================
def compile_stack(
children: list[dict[str, Any]],
*,
direction: str = "vertical",
gap: int = 12,
appearance: str = "plain",
align: str | None = None,
justify: str | None = None,
wrap: bool | None = None,
status: str | None = None,
node_id: str | None = None,
) -> dict[str, Any]:
return _compact(
{
"type": "stack",
"id": node_id,
"direction": direction,
"gap": gap,
"appearance": appearance,
"align": align,
"justify": justify,
"wrap": wrap,
"status": status,
"children": children,
}
)
def compile_grid(
children: list[dict[str, Any]],
*,
columns: int,
gap: int = 12,
appearance: str = "plain",
status: str | None = None,
node_id: str | None = None,
) -> dict[str, Any]:
return _compact(
{
"type": "grid",
"id": node_id,
"columns": columns,
"gap": gap,
"appearance": appearance,
"status": status,
"children": children,
}
)
def compile_card(
children: list[dict[str, Any]], *, status: str | None = None
) -> dict[str, Any]:
return compile_stack(
children,
direction="vertical",
gap=12,
appearance="card",
status=status,
)
def compile_section(
*,
title: str | None = None,
description: str | None = None,
icon: UiHintIcon | None = None,
children: list[dict[str, Any]],
status: str | None = None,
) -> dict[str, Any]:
section_children: list[dict[str, Any]] = []
header_row_children: list[dict[str, Any]] = []
if icon:
header_row_children.append(compile_icon_node(icon))
if title:
header_row_children.append(compile_text(title, role="title"))
if header_row_children:
section_children.append(
compile_stack(
header_row_children,
direction="horizontal",
gap=8,
align="center",
)
)
if description:
section_children.append(compile_text(description, role="caption"))
section_children.extend(children)
return compile_stack(
section_children,
direction="vertical",
gap=12,
appearance="section",
status=status,
)
# ============================================================
# Block Compilation
# ============================================================
def compile_action_row(actions: list[UiHintAction]) -> dict[str, Any] | None:
if not actions:
return None
buttons = [compile_button(a) for a in actions]
return compile_stack(
buttons,
direction="horizontal",
gap=8,
align="center",
wrap=True,
)
def compile_list_item(item: UiHintListItem) -> dict[str, Any]:
lead_children: list[dict[str, Any]] = [compile_text(item.title, role="body")]
if item.subtitle:
lead_children.append(compile_text(item.subtitle, role="caption"))
if item.description:
lead_children.append(compile_text(item.description, role="caption"))
lead_block = compile_stack(
lead_children,
direction="vertical",
gap=4,
)
main_row_children: list[dict[str, Any]] = []
if item.icon:
main_row_children.append(compile_icon_node(item.icon))
main_row_children.append(lead_block)
row = compile_stack(
main_row_children,
node_id=item.id,
direction="horizontal",
gap=8,
align="center",
)
trailing_children: list[dict[str, Any]] = []
if item.status:
trailing_children.append(
compile_badge(
label=_status_label(item.status.value),
status=item.status.value,
)
)
if trailing_children:
header = compile_stack(
[row, compile_stack(trailing_children, direction="horizontal", gap=8)],
direction="horizontal",
gap=8,
justify="space-between",
align="center",
)
else:
header = row
children: list[dict[str, Any]] = [header]
action_row = compile_action_row(item.actions)
if action_row:
children.append(action_row)
return compile_stack(
children,
direction="vertical",
gap=8,
appearance="card" if item.actions or item.description else "plain",
status=item.status.value if item.status else None,
node_id=item.id,
)
def compile_list_block(items: list[UiHintListItem]) -> dict[str, Any]:
return compile_stack(
[compile_list_item(item) for item in items],
direction="vertical",
gap=8,
)
def compile_section_block(
section: UiHintSection, default_status: str
) -> dict[str, Any]:
"""
修复点:
- 不再先把 title/description 塞进 children 再切片
- 避免 description 重复输出
"""
body_children: list[dict[str, Any]] = []
if section.content:
body_children.append(
compile_text(
section.content,
role="body",
format=section.content_format,
)
)
if section.items:
body_children.append(compile_kv(section.items))
if section.list_items:
body_children.append(compile_list_block(section.list_items))
action_row = compile_action_row(section.actions)
if action_row:
body_children.append(action_row)
if section.title or section.description or section.icon:
return compile_section(
title=section.title,
description=section.description,
icon=section.icon,
children=body_children,
status=default_status,
)
return compile_stack(
body_children,
direction="vertical",
gap=12,
)
# ============================================================
# Top-level Compilation
# ============================================================
def compile_header(hints: UiHintsPayload) -> dict[str, Any] | None:
status = compile_status(hints.status)
title_row_children: list[dict[str, Any]] = []
if hints.icon:
title_row_children.append(compile_icon_node(hints.icon))
if hints.title:
title_row_children.append(compile_text(hints.title, role="title"))
right_children: list[dict[str, Any]] = []
if _status_badge_needed(hints.intent, hints.status):
right_children.append(compile_badge(_status_label(status), status))
header_children: list[dict[str, Any]] = []
if title_row_children and right_children:
header_children.append(
compile_stack(
[
compile_stack(
title_row_children,
direction="horizontal",
gap=8,
align="center",
),
compile_stack(
right_children,
direction="horizontal",
gap=8,
align="center",
),
],
direction="horizontal",
gap=8,
justify="space-between",
align="center",
)
)
elif title_row_children:
header_children.append(
compile_stack(
title_row_children,
direction="horizontal",
gap=8,
align="center",
)
)
elif right_children:
header_children.append(
compile_stack(
right_children,
direction="horizontal",
gap=8,
align="center",
)
)
if hints.description:
header_children.append(compile_text(hints.description, role="caption"))
if not header_children:
return None
return compile_stack(
header_children,
direction="vertical",
gap=8,
)
def compile_body_blocks(hints: UiHintsPayload) -> list[dict[str, Any]]:
blocks: list[dict[str, Any]] = []
if hints.body:
blocks.append(
compile_text(
hints.body,
role="body",
format=hints.body_format,
status=compile_status(hints.status)
if hints.intent == UiHintIntent.STATUS
else None,
)
)
if hints.items:
blocks.append(compile_kv(hints.items))
if hints.list_items:
blocks.append(compile_list_block(hints.list_items))
if hints.sections:
blocks.extend(
[
compile_section_block(section, compile_status(hints.status))
for section in hints.sections
]
)
return blocks
def compile_footer(hints: UiHintsPayload) -> dict[str, Any] | None:
return compile_action_row(hints.actions)
def _root_appearance(intent: UiHintIntent) -> str:
if intent in {UiHintIntent.DATA, UiHintIntent.STATUS, UiHintIntent.MIXED}:
return "card"
if intent == UiHintIntent.FORM:
return "section"
return "plain"
def _root_gap(intent: UiHintIntent) -> int:
if intent == UiHintIntent.FORM:
return 16
return 12
def compile_root(hints: UiHintsPayload) -> dict[str, Any]:
"""
intent 只影响默认布局风格，不决定字段是否保留。
"""
children = _non_empty(
[
compile_header(hints),
*compile_body_blocks(hints),
compile_footer(hints),
]
)
if not children:
children = [compile_text("No content", role="body")]
return compile_stack(
children,
direction="vertical",
gap=_root_gap(hints.intent),
appearance=_root_appearance(hints.intent),
status=compile_status(hints.status),
)
# ============================================================
# Public API
# ============================================================
def compile(hints: UiHintsPayload) -> dict[str, Any]:
"""
将描述性 UiHints 编译为渲染性 UiSchemaRenderer。
保证:
- title / description / body / items / listItems / sections / actions / icon 尽量保留
- intent 只影响默认包装风格
- meta 中常用 requestId/toolId/traceId/userId 会透传
"""
root = compile_root(hints)
meta_keys = ("requestId", "toolId", "traceId", "userId")
meta = {k: hints.meta.get(k) for k in meta_keys if hints.meta.get(k) is not None}
result: dict[str, Any] = {
"version": "2.0",
"locale": "zh-CN",
"status": compile_status(hints.status),
"theme": "default",
"root": root,
}
if meta:
result["meta"] = meta
return _compact(result)