Текстовый файл из учебного комплекта, не свидетельство установленного runtime. Вернуться к учебнику · Исходный файл
1 from __future__ import annotations 2 3 import json 4 from hashlib import sha256 5 from typing import Any 6 7 from src_agent.utils.nats_logger import NatsLogger 8 from src_shared.contracts import AgentInboundRequest, WorkflowRunResponse 9 10 11 class UiEventPublisher: 12 def __init__(self, nats_logger: NatsLogger) -> None: 13 self.nats_logger = nats_logger 14 15 async def publish(self, req: AgentInboundRequest, resp: WorkflowRunResponse) -> None: 16 trace = { 17 "trace_id": str(req.trace_id), 18 "correlation_id": str(req.correlation_id or req.trace_id), 19 "request_id": str(req.request_id), 20 "session_id": str(resp.session_id) if resp.session_id else None, 21 } 22 23 sent: set[str] = set() 24 thought_handlers: list[dict[str, Any]] = [] 25 regular_handlers: list[dict[str, Any]] = [] 26 27 for ev in _normalize_client_events(resp.client_events): 28 (thought_handlers if _is_thought_handler(ev) else regular_handlers).append(ev) 29 30 if resp.client_handler: 31 (thought_handlers if _is_thought_handler(resp.client_handler) else regular_handlers).append(resp.client_handler) 32 33 for handler in thought_handlers: 34 await self._publish_client_command(handler, trace, sent) 35 36 msg_text = _extract_result_message(resp) 37 if msg_text and not _has_duplicate_client_message(resp, msg_text): 38 await self.nats_logger.log( 39 "command", 40 "message", 41 {"type": "answer" if resp.status != "FAILED" else "system", "text": msg_text, "trace": trace}, 42 ) 43 44 for handler in regular_handlers: 45 await self._publish_client_command(handler, trace, sent) 46 47 if resp.status == "FAILED" and resp.errors and not _has_explicit_error_command(resp): 48 await self._publish_client_command( 49 { 50 "command": "SHOW_ERROR_MESSAGE", 51 "payload": {"message": resp.errors[0].message, "code": resp.errors[0].code}, 52 }, 53 trace, 54 sent, 55 ) 56 57 async def _publish_client_command( 58 self, 59 handler: dict[str, Any], 60 trace: dict[str, Any], 61 sent: set[str], 62 ) -> None: 63 fp = _fingerprint_client_command(handler) 64 if fp in sent: 65 return 66 sent.add(fp) 67 68 command = handler.get("command") 69 if not isinstance(command, str) or not command.strip(): 70 return 71 command_name = command.strip() 72 73 if _is_thought_command(command_name): 74 data = _thought_event_data(handler.get("payload"), trace) 75 if data: 76 await self.nats_logger.log("command", "thought", data) 77 return 78 79 artifacts = _payload_to_artifacts(handler.get("payload")) 80 data: dict[str, Any] = {"command": command_name, "trace": trace} 81 if artifacts: 82 data["artifacts"] = artifacts 83 await self.nats_logger.log("command", "client", data) 84 85 86 def _extract_result_message(resp: WorkflowRunResponse) -> str | None: 87 if isinstance(resp.result, str) and resp.result.strip(): 88 return resp.result.strip() 89 if resp.errors: 90 first = resp.errors[0] 91 if isinstance(first.message, str) and first.message.strip(): 92 return first.message.strip() 93 return None 94 95 96 def _normalize_client_events(events: Any) -> list[dict[str, Any]]: 97 if not isinstance(events, list): 98 return [] 99 return [e for e in events if isinstance(e, dict)] 100 101 102 def _has_duplicate_client_message(resp: WorkflowRunResponse, msg_text: str) -> bool: 103 target = msg_text.strip() 104 if not target: 105 return False 106 if _client_command_message_text(resp.client_handler) == target: 107 return True 108 for ev in _normalize_client_events(resp.client_events): 109 if _client_command_message_text(ev) == target: 110 return True 111 return False 112 113 114 def _client_command_message_text(handler: Any) -> str | None: 115 if not isinstance(handler, dict): 116 return None 117 command = handler.get("command") 118 if not isinstance(command, str): 119 return None 120 if command.strip().upper() not in {"SHOW_MESSAGE", "ASK_USER_INPUT", "SHOW_ERROR_MESSAGE"}: 121 return None 122 payload = handler.get("payload") 123 if not isinstance(payload, dict): 124 return None 125 for key in ("summary", "prompt", "message"): 126 value = payload.get(key) 127 if isinstance(value, str) and value.strip(): 128 return value.strip() 129 return None 130 131 132 def _has_explicit_error_command(resp: WorkflowRunResponse) -> bool: 133 def _is_error_cmd(cmd: Any) -> bool: 134 return isinstance(cmd, str) and cmd.strip().upper().startswith("SHOW_ERROR") 135 136 if resp.client_handler and _is_error_cmd(resp.client_handler.get("command")): 137 return True 138 return any(_is_error_cmd(ev.get("command")) for ev in _normalize_client_events(resp.client_events)) 139 140 141 def _is_thought_command(command: str) -> bool: 142 return command.strip().upper() in {"SHOW_THOUGHT", "EMIT_THOUGHT"} 143 144 145 def _is_thought_handler(handler: Any) -> bool: 146 return isinstance(handler, dict) and isinstance(handler.get("command"), str) and _is_thought_command(handler["command"]) 147 148 149 def _thought_event_data(payload: Any, trace: dict[str, Any]) -> dict[str, Any] | None: 150 if not isinstance(payload, dict): 151 return None 152 summary = payload.get("summary") 153 content = payload.get("content") 154 scenario = payload.get("scenario") 155 tools = payload.get("tools") 156 157 data: dict[str, Any] = {"summary": "Thinking…", "trace": trace} 158 if isinstance(summary, str) and summary.strip(): 159 data["summary"] = summary.strip() 160 if isinstance(content, str) and content.strip(): 161 data["content"] = content.strip() 162 if isinstance(scenario, dict): 163 sid = scenario.get("id") 164 reason = scenario.get("reason") 165 out_scenario: dict[str, Any] = {} 166 if isinstance(sid, str) and sid.strip(): 167 out_scenario["id"] = sid.strip() 168 if isinstance(reason, str) and reason.strip(): 169 out_scenario["reason"] = reason.strip() 170 if out_scenario: 171 data["scenario"] = out_scenario 172 if isinstance(tools, list): 173 tool_names = [t.strip() for t in tools if isinstance(t, str) and t.strip()] 174 if tool_names: 175 data["tools"] = tool_names 176 return data 177 178 179 def _fingerprint_client_command(handler: dict[str, Any]) -> str: 180 try: 181 raw = json.dumps(handler, sort_keys=True, ensure_ascii=True, separators=(",", ":")) 182 except Exception: 183 raw = repr(handler) 184 return sha256(raw.encode("utf-8")).hexdigest() 185 186 187 def _payload_to_artifacts(payload: Any) -> dict[str, Any] | None: 188 if not isinstance(payload, dict) or not payload: 189 return None 190 return {"all": ["ui"], "last": "ui", "payload": {"ui": payload}}