feat: 工具组特性回流九问runtime-agent-runtime-AtomGit

M.gitignore+3-0

	@@ -38,6 +38,9 @@ Thumbs.db
38	logs	38	logs
39	*.log	39	*.log
40		40
		41	+# Python venv
		42	+.venv311/
		43	+
41	# Runtime data	44	# Runtime data
42	pids	45	pids
43	*.pid	46	*.pid

Aapplications/a2a_service/.env.example+112-0

@@ -0,0 +1,112 @@
	1	+# ── 配置项加解密────────────────
	2	+# 未设置时以下变量均为明文；设置后可将敏感值换为 script/crypto 生成的密文。
	3	+# 生成主密钥: python applications/a2a_service/script/crypto/generate_master_key.py
	4	+# 加密字符串: python applications/a2a_service/script/crypto/crypto_cli.py --encrypt "明文"
	5	+# 注意: ir_execution_service 使用的 SERVER_AES_MASTER_KEY 用于「内存加密」，与
	6	+# 本处的 AES_MASTER_KEY（配置字段加解密）用途不同，请勿混用。
	7	+AES_MASTER_KEY=
	8	+
	9	+# ── App ──────────────────────────────────────────────────────────────────────
	10	+APP_NAME=A2A Service
	11	+
	12	+# ── Redis（会话状态）─────────────────────────────────────────────────────────
	13	+REDIS_HOST=localhost
	14	+REDIS_PORT=6379
	15	+REDIS_DB=0
	16	+REDIS_PASSWORD=
	17	+REDIS_SESSION_TTL=180
	18	+REDIS_CHECKPOINTER_TTL_MINUTES=60
	19	+
	20	+# ── Bootstrap 协调（Redis 锁 + 状态协同，多实例并发启动时使用）────────────
	21	+BOOTSTRAP_COORDINATION_ENABLED=False
	22	+BOOTSTRAP_LOCK_NAME=a2a_global_bootstrap
	23	+BOOTSTRAP_LOCK_TTL_SEC=180
	24	+BOOTSTRAP_WAIT_TIMEOUT_SEC=300
	25	+BOOTSTRAP_POLL_INTERVAL_SEC=1.0
	26	+
	27	+# ── Rate Limit（入口限流）─────────────────────────────────────────────────
	28	+RATE_LIMIT_MAX_REQUESTS=5
	29	+RATE_LIMIT_WINDOW_SECONDS=120
	30	+GLOBAL_RATE_LIMIT_MAX_REQUESTS=100
	31	+GLOBAL_RATE_LIMIT_WINDOW_SECONDS=30
	32	+
	33	+# ── VersatileAdapter（内部 A2A 服务地址）─────────────────────────────────────
	34	+VERSATILE_ADAPTER_URL=http://localhost:8091
	35	+
	36	+# ── FastAPI ───────────────────────────────────────────────────────────────────
	37	+FASTAPI_HOST=0.0.0.0
	38	+FASTAPI_PORT=8090
	39	+FASTAPI_DEBUG=False
	40	+FASTAPI_WORKERS=1
	41	+
	42	+# ── Log ───────────────────────────────────────────────────────────────────────
	43	+LOG_LEVEL=INFO
	44	+LOG_DIR=logs
	45	+
	46	+# ── LLM（上游保留字段，供其他 Agent 参考；EDPAgent 不读此组）──────────────
	47	+LLM_PROVIDER=llm_OpenAI
	48	+LLM_API_BASE=https://api.deepseek.com
	49	+LLM_API_KEY=
	50	+LLM_MODEL_NAME=deepseek-chat
	51	+LLM_VERIFY_SSL=False
	52	+LLM_USER_ID=
	53	+LLM_TOKEN=
	54	+
	55	+# ── LLM（EDPAgent 实际读取，统一 PLANNING_AGENT_MODEL_* 命名）──────────────
	56	+# EDPAgent/config.py 只读这一组。与上面的 LLM_* 功能重叠时，
	57	+# PLANNING_AGENT_MODEL_* 优先；建议新部署只填这一组，LLM_* 留空。
	58	+PLANNING_AGENT_MODEL_NAME=
	59	+PLANNING_AGENT_MODEL_API_KEY=
	60	+PLANNING_AGENT_MODEL_BASE_URL=
	61	+PLANNING_AGENT_MODEL_TIMEOUT=120
	62	+SKILL_LLM_TLS_VERIFY=false
	63	+
	64	+# ── 企业内网 LLM 鉴权（可选，按所用网关要求填写）──────────────────────────
	65	+# 每个鉴权字段由「值变量」和「header 名变量」组成一对，均需填写。
	66	+# ⚠️ 值变量有内容时，对应的 header 名变量必须同时填写，否则启动报错。
	67	+#
	68	+# 示例 A — 通用网关（header 名为 token / userId）：
	69	+# PLANNING_AGENT_MODEL_TOKEN=my-token
	70	+# PLANNING_AGENT_MODEL_TOKEN_HEADER=token
	71	+# PLANNING_AGENT_MODEL_USER_ID=user-001
	72	+# PLANNING_AGENT_MODEL_USER_ID_HEADER=userId
	73	+#
	74	+# 示例 B — 银行网关（X-Bank-Token / X-User-Id，附加渠道号）：
	75	+# PLANNING_AGENT_MODEL_TOKEN=gw-token-001
	76	+# PLANNING_AGENT_MODEL_TOKEN_HEADER=X-Bank-Token
	77	+# PLANNING_AGENT_MODEL_USER_ID=cust-1001
	78	+# PLANNING_AGENT_MODEL_USER_ID_HEADER=X-User-Id
	79	+# PLANNING_AGENT_MODEL_EXTRA_HEADERS={"X-Channel-Id":"mobile-banking"}
	80	+#
	81	+PLANNING_AGENT_MODEL_TOKEN=
	82	+PLANNING_AGENT_MODEL_TOKEN_HEADER=
	83	+PLANNING_AGENT_MODEL_USER_ID=
	84	+PLANNING_AGENT_MODEL_USER_ID_HEADER=
	85	+PLANNING_AGENT_MODEL_EXTRA_HEADERS=
	86	+
	87	+# ── DPA Agent ─────────────────────────────────────────────────────────────────
	88	+DPA_AGENT_ID=edp_agent
	89	+DPA_AGENT_NAME=EDP Agent
	90	+DPA_MAX_ITERATIONS=30
	91	+
	92	+# ── Versatile 工作流最终结果节点名（a2a_service/config.py 读取）───────────
	93	+va_workflow_result_node=
	94	+
	95	+# ── 沙箱配置 ──
	96	+# 沙箱服务地址
	97	+SANDBOX_URL=
	98	+# skill 在沙箱中的存储位置。
	99	+SKILL_TARGET_PATH=/tmp
	100	+
	101	+# MCP相关配置信息
	102	+
	103	+# 鉴权Token信息
	104	+MCP_ACCESS_TOKEN=
	105	+# MCP名称
	106	+MCP_APP_NAME=
	107	+# MCP地址
	108	+MCP_MASTER_URL=
	109	+# MCP备用地址
	110	+MCP_STANDBY_URL=
	111	+# MCP超时
	112	+MCP_TIMEOUT=

Aapplications/a2a_service/agents/__init__.py+9-0

@@ -0,0 +1,9 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+"""
	5	+agents 包 — 包含所有智能体实现。
	6	+
	7	+当前包含：
	8	+ - EDPAgent: 企业动态规划智能体
	9	+"""

Aapplications/a2a_service/app.py+458-0

@@ -0,0 +1,458 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+"""
	5	+A2A Service 进程入口。
	6	+
	7	+暴露端点：
	8	+ POST /v1/{project_id}/agents/{agent_id}/conversations/{conv_id} — 定制化 Versatile 入口
	9	+ GET /a2a/.well-known/agent-card.json — A2A 标准 Agent Card
	10	+ POST /a2a/ — A2A 标准 JSON-RPC 入口
	11	+
	12	+两条路径共用同一个 Executor + RedisTaskStore，Task 状态一致。
	13	+"""
	14	+
	15	+from __future__ import annotations
	16	+
	17	+import asyncio
	18	+import os
	19	+import socket
	20	+import sys
	21	+import time
	22	+from contextlib import asynccontextmanager
	23	+from typing import Any, Optional
	24	+
	25	+import httpx
	26	+from a2a.client import ClientConfig, ClientFactory
	27	+from a2a.server.request_handlers import DefaultRequestHandler
	28	+from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
	29	+from a2a.types.a2a_pb2 import AgentCapabilities, AgentCard, AgentInterface
	30	+from a2a.utils.constants import PROTOCOL_VERSION_1_0, TransportProtocol
	31	+from fastapi import FastAPI
	32	+from loguru import logger
	33	+from starlette.applications import Starlette
	34	+
	35	+from agents.EDPAgent import initialize
	36	+from common.redis_client import RedisClient
	37	+from common.redis_task_store import RedisTaskStore
	38	+from config import get_settings
	39	+from orchestrator.executor import Executor
	40	+from orchestrator.user_router import router as user_router
	41	+from tools.simulate_router.simulate import router as simulate_router
	42	+
	43	+os.environ['NO_PROXY'] = 'localhost,127.0.0.1'
	44	+
	45	+
	46	+def setup_logging() -> None:
	47	+ settings = get_settings()
	48	+
	49	+ logger.remove()
	50	+
	51	+ logger.add(
	52	+ sys.stderr,
	53	+ level=settings.log_level.upper() if settings.log_level else "INFO",
	54	+ format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> \| "
	55	+ "<level>{level: <8}</level> \| "
	56	+ "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> \| "
	57	+ "<level>{message}</level>"
	58	+ )
	59	+
	60	+ if settings.log_dir:
	61	+ log_dir = settings.log_dir
	62	+ if log_dir:
	63	+ os.makedirs(log_dir, exist_ok=True)
	64	+ log_file_with_pid = f"{log_dir}{os.sep}/process_{os.getpid()}.log"
	65	+ audit_log = f"{log_dir}{os.sep}/audit_{os.getpid()}.log"
	66	+ logger.add(
	67	+ log_file_with_pid,
	68	+ level="INFO",
	69	+ rotation="20 MB",
	70	+ retention="7 days",
	71	+ compression="gz",
	72	+ format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> \x01 "
	73	+ "<level>{level: <8}</level> \x01 "
	74	+ "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> \x01 "
	75	+ "<level>{message}</level>",
	76	+ filter=lambda record: len(record["extra"]) == 0
	77	+ )
	78	+
	79	+ logger.add(
	80	+ log_file_with_pid,
	81	+ level="INFO",
	82	+ format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> \x01 "
	83	+ "<level>{level: <8}</level> \x01 "
	84	+ "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> \x01 "
	85	+ "<cyan>{extra[trace_id]}</cyan> \x01 "
	86	+ "<cyan>{extra[agent_id]}</cyan> \x01 "
	87	+ "<cyan>{extra[conversation_id]}</cyan> \x01 "
	88	+ "<level>{message}</level>",
	89	+ filter=lambda record: len(record["extra"]) > 0 and "tag" not in record["extra"] and "source" not in record[
	90	+ "extra"]
	91	+ )
	92	+
	93	+ logger.add(
	94	+ log_file_with_pid,
	95	+ level="INFO",
	96	+ format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> \x01 "
	97	+ "<level>{level: <8}</level> \x01 "
	98	+ "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> \x01 "
	99	+ "<cyan>{extra[trace_id]}</cyan> \x01 "
	100	+ "<cyan>{extra[agent_id]}</cyan> \x01 "
	101	+ "<cyan>{extra[conversation_id]}</cyan> \x01 "
	102	+ "<cyan>{extra[tag]}</cyan> \x01 "
	103	+ "<cyan>{extra[cost]}</cyan> \x01 "
	104	+ "<level>{message}</level>",
	105	+ filter=lambda record: "tag" in record["extra"]
	106	+ )
	107	+ logger.add(
	108	+ audit_log,
	109	+ level="INFO",
	110	+ rotation="20 MB",
	111	+ retention="30 days",
	112	+ compression="gz",
	113	+ format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> \x01 "
	114	+ "<level>{level: <8}</level> \x01 "
	115	+ "<cyan>{extra[source]}</cyan> \x01 "
	116	+ "<cyan>{extra[user]}</cyan> \x01 "
	117	+ "<cyan>{extra[result]}</cyan> \x01 "
	118	+ "<cyan>{extra[terminal]}</cyan> \x01 "
	119	+ "<level>{message}</level>",
	120	+ filter=lambda record: "source" in record["extra"]
	121	+ )
	122	+
	123	+
	124	+setup_logging()
	125	+
	126	+
	127	+def _build_va_card(url: str) -> AgentCard:
	128	+ card = AgentCard(
	129	+ name="VersatileAdapter",
	130	+ description="Versatile 低代码平台 A2A 适配器",
	131	+ version="1.0.0",
	132	+ )
	133	+ card.supported_interfaces.append(
	134	+ AgentInterface(
	135	+ protocol_binding=TransportProtocol.JSONRPC,
	136	+ url=url,
	137	+ protocol_version=PROTOCOL_VERSION_1_0,
	138	+ )
	139	+ )
	140	+ card.capabilities.CopyFrom(AgentCapabilities(streaming=True))
	141	+ return card
	142	+
	143	+
	144	+def _build_dpa_card() -> AgentCard:
	145	+ settings = get_settings()
	146	+ host = settings.fastapi_host or "localhost"
	147	+ if host == "0.0.0.0":
	148	+ host = "localhost"
	149	+ port = settings.fastapi_port or 8090
	150	+ url = f"http://{host}:{port}/a2a/"
	151	+ card = AgentCard(
	152	+ name="DPA Service",
	153	+ description="EDPA 编排服务：规划并委托 VersatileAdapter 执行子任务",
	154	+ version="1.0.0",
	155	+ )
	156	+ card.supported_interfaces.append(
	157	+ AgentInterface(
	158	+ protocol_binding=TransportProtocol.JSONRPC,
	159	+ url=url,
	160	+ protocol_version=PROTOCOL_VERSION_1_0,
	161	+ )
	162	+ )
	163	+ card.capabilities.CopyFrom(AgentCapabilities(streaming=True))
	164	+ return card
	165	+
	166	+
	167	+def _bootstrap_lock_key(lock_name: str) -> str:
	168	+ return f"a2a:bootstrap:lock:{lock_name}"
	169	+
	170	+
	171	+def _bootstrap_status_key(lock_name: str) -> str:
	172	+ return f"a2a:bootstrap:status:{lock_name}"
	173	+
	174	+
	175	+async def _set_bootstrap_status(
	176	+ redis: RedisClient,
	177	+ *,
	178	+ status_key: str,
	179	+ status: str,
	180	+ owner_id: str,
	181	+ message: Optional[str],
	182	+ ttl_seconds: int,
	183	+) -> None:
	184	+ payload: dict[str, Any] = {
	185	+ "status": status,
	186	+ "owner_id": owner_id,
	187	+ "message": message,
	188	+ "update_time": int(time.time()),
	189	+ }
	190	+ await redis.set_json(status_key, payload, ex=max(int(ttl_seconds), 60))
	191	+
	192	+
	193	+async def _wait_for_bootstrap_ready(
	194	+ redis: RedisClient,
	195	+ *,
	196	+ status_key: str,
	197	+ timeout_seconds: int,
	198	+ poll_interval_seconds: float,
	199	+) -> bool:
	200	+ deadline = time.time() + max(int(timeout_seconds), 1)
	201	+ poll = max(float(poll_interval_seconds), 0.2)
	202	+ attempts = 0
	203	+ while time.time() < deadline:
	204	+ attempts += 1
	205	+ state = await redis.get_json(status_key) or {}
	206	+ status = str(state.get("status") or "").lower()
	207	+ if attempts == 1 or attempts % 10 == 0:
	208	+ remaining = max(int(deadline - time.time()), 0)
	209	+ logger.info(
	210	+ "[A2AService] FOLLOWER 等待 bootstrap: attempt={}, status={}, owner={}, remaining={}s",
	211	+ attempts,
	212	+ status or "<empty>",
	213	+ state.get("owner_id"),
	214	+ remaining,
	215	+ )
	216	+ if status == "ready":
	217	+ return True
	218	+ if status == "failed":
	219	+ logger.error("[A2AService] FOLLOWER 检测到 bootstrap 失败: {}", state)
	220	+ return False
	221	+ await asyncio.sleep(poll)
	222	+ return False
	223	+
	224	+
	225	+async def _run_global_bootstrap_once() -> None:
	226	+ logger.info("[A2AService] LEADER 全局 bootstrap 无额外任务，标记为 ready")
	227	+
	228	+
	229	+class _BootstrapCoordinator:
	230	+ """封装 bootstrap 协调流程，降低 lifespan 复杂度。"""
	231	+
	232	+ def __init__(self, *, settings: Any, redis: RedisClient) -> None:
	233	+ self.settings = settings
	234	+ self.redis = redis
	235	+
	236	+ self.bootstrap_enabled = bool(getattr(settings, "bootstrap_coordination_enabled", False))
	237	+ self.bootstrap_lock_name = getattr(settings, "bootstrap_lock_name", "a2a_global_bootstrap")
	238	+ self.bootstrap_owner_id = f"{socket.gethostname()}-{os.getpid()}"
	239	+ self.bootstrap_lock_key = _bootstrap_lock_key(self.bootstrap_lock_name)
	240	+ self.bootstrap_status_key = _bootstrap_status_key(self.bootstrap_lock_name)
	241	+ self.bootstrap_lock_ttl = max(int(getattr(settings, "bootstrap_lock_ttl_sec", 180)), 1)
	242	+ self.bootstrap_wait_timeout = max(int(getattr(settings, "bootstrap_wait_timeout_sec", 300)), 1)
	243	+ self.bootstrap_poll_interval = max(float(getattr(settings, "bootstrap_poll_interval_sec", 1.0)), 0.2)
	244	+
	245	+ # ready 状态用于后启动实例快速放行，TTL 设长一些以避免 leader 退出后状态过早失效。
	246	+ self.bootstrap_status_ttl = max(self.bootstrap_wait_timeout * 2, 1800)
	247	+
	248	+ self.leader_locked = False
	249	+ self.bootstrap_ready = False
	250	+
	251	+ async def run(self) -> None:
	252	+ if not self.bootstrap_enabled:
	253	+ logger.info("[A2AService] 已禁用 bootstrap 协调，跳过 Redis leader/follower 编排")
	254	+ return
	255	+
	256	+ logger.info(
	257	+ "[A2AService] bootstrap 启动参数: lock_name={}, owner={}, ttl={}s, wait_timeout={}s, poll_interval={}s",
	258	+ self.bootstrap_lock_name,
	259	+ self.bootstrap_owner_id,
	260	+ self.bootstrap_lock_ttl,
	261	+ self.bootstrap_wait_timeout,
	262	+ self.bootstrap_poll_interval,
	263	+ )
	264	+
	265	+ self.leader_locked = await self.redis.acquire_lock(
	266	+ lock_key=self.bootstrap_lock_key,
	267	+ owner_id=self.bootstrap_owner_id,
	268	+ ttl_seconds=self.bootstrap_lock_ttl,
	269	+ )
	270	+
	271	+ if self.leader_locked:
	272	+ await self._run_leader_flow()
	273	+ else:
	274	+ await self._run_follower_flow()
	275	+
	276	+ async def mark_failed_if_needed(self, exc: Exception) -> None:
	277	+ if self.bootstrap_enabled and self.leader_locked and not self.bootstrap_ready:
	278	+ try:
	279	+ await _set_bootstrap_status(
	280	+ self.redis,
	281	+ status_key=self.bootstrap_status_key,
	282	+ status="failed",
	283	+ owner_id=self.bootstrap_owner_id,
	284	+ message=str(exc),
	285	+ ttl_seconds=self.bootstrap_status_ttl,
	286	+ )
	287	+ except Exception as mark_exc:
	288	+ logger.debug(
	289	+ "[A2AService] 标记 bootstrap failed 失败（忽略）: {}",
	290	+ mark_exc,
	291	+ )
	292	+
	293	+ async def close(self) -> None:
	294	+ await self._release_leader_lock(reason="service closing")
	295	+
	296	+ async def _run_leader_flow(self) -> None:
	297	+ await _set_bootstrap_status(
	298	+ self.redis,
	299	+ status_key=self.bootstrap_status_key,
	300	+ status="initializing",
	301	+ owner_id=self.bootstrap_owner_id,
	302	+ message="leader is running one-time bootstrap tasks",
	303	+ ttl_seconds=self.bootstrap_status_ttl,
	304	+ )
	305	+
	306	+ try:
	307	+ await _run_global_bootstrap_once()
	308	+ except Exception as bootstrap_exc:
	309	+ await _set_bootstrap_status(
	310	+ self.redis,
	311	+ status_key=self.bootstrap_status_key,
	312	+ status="failed",
	313	+ owner_id=self.bootstrap_owner_id,
	314	+ message=str(bootstrap_exc),
	315	+ ttl_seconds=self.bootstrap_status_ttl,
	316	+ )
	317	+ raise
	318	+
	319	+ await _set_bootstrap_status(
	320	+ self.redis,
	321	+ status_key=self.bootstrap_status_key,
	322	+ status="ready",
	323	+ owner_id=self.bootstrap_owner_id,
	324	+ message="one-time bootstrap finished",
	325	+ ttl_seconds=self.bootstrap_status_ttl,
	326	+ )
	327	+ self.bootstrap_ready = True
	328	+ logger.info(
	329	+ "[A2AService] 节点角色=LEADER，bootstrap 完成: lock={}, owner={}",
	330	+ self.bootstrap_lock_name,
	331	+ self.bootstrap_owner_id,
	332	+ )
	333	+ await self._release_leader_lock(reason="bootstrap finished")
	334	+
	335	+ async def _run_follower_flow(self) -> None:
	336	+ logger.info(
	337	+ "[A2AService] 节点角色=FOLLOWER，等待 LEADER bootstrap 完成: lock={}, owner={}",
	338	+ self.bootstrap_lock_name,
	339	+ self.bootstrap_owner_id,
	340	+ )
	341	+ ready = await _wait_for_bootstrap_ready(
	342	+ self.redis,
	343	+ status_key=self.bootstrap_status_key,
	344	+ timeout_seconds=self.bootstrap_wait_timeout,
	345	+ poll_interval_seconds=self.bootstrap_poll_interval,
	346	+ )
	347	+ if not ready:
	348	+ state = await self.redis.get_json(self.bootstrap_status_key)
	349	+ raise RuntimeError(f"等待 LEADER bootstrap 完成失败，状态: {state}")
	350	+ logger.info("[A2AService] FOLLOWER 检测到 bootstrap ready")
	351	+
	352	+ async def _release_leader_lock(self, *, reason: str) -> None:
	353	+ if not (self.bootstrap_enabled and self.leader_locked):
	354	+ return
	355	+ try:
	356	+ released = await self.redis.release_lock(
	357	+ lock_key=self.bootstrap_lock_key,
	358	+ owner_id=self.bootstrap_owner_id,
	359	+ )
	360	+ if released:
	361	+ logger.info(
	362	+ "[A2AService] LEADER 已释放 bootstrap 锁: lock={}, owner={}, reason={}",
	363	+ self.bootstrap_lock_name,
	364	+ self.bootstrap_owner_id,
	365	+ reason,
	366	+ )
	367	+ except Exception as release_exc:
	368	+ logger.warning("[A2AService] 释放 bootstrap 锁异常: {}", release_exc)
	369	+ finally:
	370	+ self.leader_locked = False
	371	+
	372	+
	373	+
	374	+@asynccontextmanager
	375	+async def lifespan(fastapi_app: FastAPI):
	376	+ settings = get_settings()
	377	+ redis = RedisClient()
	378	+ http_client: Optional[httpx.AsyncClient] = None
	379	+ bootstrap = _BootstrapCoordinator(settings=settings, redis=redis)
	380	+
	381	+ try:
	382	+ await redis.connect(settings.redis_url)
	383	+ await bootstrap.run()
	384	+
	385	+ await initialize()
	386	+ logger.info("[A2AService] Agent 初始化完成")
	387	+
	388	+ http_client = httpx.AsyncClient()
	389	+ va_card = _build_va_card(settings.versatile_adapter_url)
	390	+ factory = ClientFactory(ClientConfig(httpx_client=http_client))
	391	+ va_client = factory.create(va_card)
	392	+
	393	+ task_store = RedisTaskStore(redis, ttl=settings.redis_session_ttl or _TTL)
	394	+ executor = Executor(va_client=va_client, redis=redis, task_store=task_store)
	395	+
	396	+ dpa_card = _build_dpa_card()
	397	+ request_handler = DefaultRequestHandler(
	398	+ agent_executor=executor,
	399	+ task_store=task_store,
	400	+ agent_card=dpa_card,
	401	+ )
	402	+
	403	+ fastapi_app.state.redis = redis
	404	+ fastapi_app.state.task_store = task_store
	405	+ fastapi_app.state.executor = executor
	406	+
	407	+ a2a_routes = create_agent_card_routes(dpa_card) + create_jsonrpc_routes(
	408	+ request_handler, rpc_url="/"
	409	+ )
	410	+ fastapi_app.mount("/a2a", Starlette(routes=a2a_routes))
	411	+
	412	+ logger.info(
	413	+ f"[A2AService] 启动完成："
	414	+ f"VersatileAdapter={settings.versatile_adapter_url}, "
	415	+ f"A2A endpoint=http://{settings.fastapi_host or '0.0.0.0'}:{settings.fastapi_port or 8090}/a2a/"
	416	+ )
	417	+ yield
	418	+
	419	+ except Exception as e:
	420	+ logger.error(f"[A2AService] 启动/运行异常: {e}")
	421	+ await bootstrap.mark_failed_if_needed(e)
	422	+ raise
	423	+
	424	+ finally:
	425	+ await bootstrap.close()
	426	+ if http_client:
	427	+ await http_client.aclose()
	428	+ await redis.disconnect()
	429	+ try:
	430	+ from openjiuwen.core.runner import Runner
	431	+ await Runner.stop()
	432	+ except Exception as stop_exc:
	433	+ logger.debug("[A2AService] Runner.stop 异常（忽略）: {}", stop_exc)
	434	+ logger.info("[A2AService] 关闭完成")
	435	+
	436	+
	437	+_TTL = 1800
	438	+
	439	+app = FastAPI(
	440	+ title="A2A Service",
	441	+ description="DPA + VersatileAdapter 编排服务，支持 Versatile 定制入口和标准 A2A 入口",
	442	+ version="1.0.0",
	443	+ lifespan=lifespan,
	444	+)
	445	+
	446	+app.include_router(user_router)
	447	+app.include_router(simulate_router)
	448	+
	449	+
	450	+@app.get("/health", tags=["Health"])
	451	+async def health_check(success: str = None):
	452	+ """服务健康检查"""
	453	+ if success is not None:
	454	+ return success
	455	+ return {
	456	+ "status": "healthy",
	457	+ "service": "A2A Service",
	458	+ }

Aapplications/a2a_service/common/__init__.py+16-0

@@ -0,0 +1,16 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+from .events import (
	5	+ AgentEvent,
	6	+ ThoughtEvent,
	7	+ AnswerEvent,
	8	+ DelegateRequest,
	9	+)
	10	+
	11	+__all__ = [
	12	+ "AgentEvent",
	13	+ "ThoughtEvent",
	14	+ "AnswerEvent",
	15	+ "DelegateRequest",
	16	+]

Aapplications/a2a_service/common/constants.py+12-0

@@ -0,0 +1,12 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+"""
	5	+Redis Key 模板函数。
	6	+"""
	7	+from __future__ import annotations
	8	+
	9	+
	10	+def session_request_key(conv_id: str) -> str:
	11	+ """首轮请求的请求头和请求体缓存。"""
	12	+ return f"session:{conv_id}:request"

Aapplications/a2a_service/common/crypto.py+171-0

@@ -0,0 +1,171 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+"""
	4	+a2a_service 侧配置敏感字段加解密（AES-GCM + HKDF-SHA256）。
	5	+
	6	+离线加解密脚本见 a2a_service/script/crypto/，与本模块算法一致。
	7	+
	8	+主密钥：环境变量 AES_MASTER_KEY（Base64 解码后须为 32 字节）。未设置时 encrypt/decrypt 对非密文透传。
	9	+"""
	10	+
	11	+from __future__ import annotations
	12	+
	13	+import base64
	14	+import logging
	15	+import os
	16	+from typing import Optional
	17	+
	18	+from Crypto.Cipher import AES
	19	+from Crypto.Hash import SHA256
	20	+from Crypto.Protocol.KDF import HKDF
	21	+from Crypto.Random import get_random_bytes
	22	+
	23	+logger = logging.getLogger(__name__)
	24	+
	25	+
	26	+def _strip_matched_quotes(text: str) -> str:
	27	+ """若字符串首尾被相同的成对单/双引号包裹，则去掉这一对引号；否则原样返回。
	28	+
	29	+ 将原本嵌入在 if 条件中的多重 `startswith/endswith` 布尔表达式抽到独立函数，
	30	+ 使主流程的判断条件更精炼。
	31	+ """
	32	+ if len(text) < 2:
	33	+ return text
	34	+ for quote in ('"', "'"):
	35	+ if text.startswith(quote) and text.endswith(quote):
	36	+ return text[1:-1]
	37	+ return text
	38	+
	39	+
	40	+def parse_master_key_from_string(key_material: Optional[str]) -> Optional[bytes]:
	41	+ """
	42	+ 将 Base64 形式的主密钥字符串解析为 32 字节（与从 AES_MASTER_KEY 环境变量读取的规则一致）。
	43	+ """
	44	+ if not key_material:
	45	+ return None
	46	+ key_base64 = _strip_matched_quotes(key_material.strip()).strip()
	47	+ if not key_base64:
	48	+ logger.error("Master key is empty after trimming.")
	49	+ return None
	50	+ padding_needed = 4 - len(key_base64) % 4
	51	+ if padding_needed != 4:
	52	+ key_base64 += "=" * padding_needed
	53	+ try:
	54	+ key_bytes = base64.b64decode(key_base64)
	55	+ except Exception as e:
	56	+ logger.error("Failed to decode master key (base64): %s", type(e).__name__)
	57	+ return None
	58	+ if len(key_bytes) != 32:
	59	+ logger.error("Invalid master key length: expected 32 bytes, got %s.", len(key_bytes))
	60	+ return None
	61	+ return key_bytes
	62	+
	63	+
	64	+class CryptoUtils:
	65	+ """AES-GCM + HKDF config encryption/decryption."""
	66	+
	67	+ def __init__(self, master_key: Optional[bytes] = None):
	68	+ if master_key is None:
	69	+ self.master_key = self._get_master_key_from_env()
	70	+ else:
	71	+ self.master_key = master_key
	72	+
	73	+ if self.master_key is not None and len(self.master_key) != 32:
	74	+ raise ValueError("master_key length must be 32 bytes")
	75	+
	76	+ @staticmethod
	77	+ def _get_master_key_from_env() -> Optional[bytes]:
	78	+ key_base64 = os.getenv("AES_MASTER_KEY")
	79	+ if not key_base64:
	80	+ logger.warning(
	81	+ "AES_MASTER_KEY is not set. Config encryption/decryption is disabled; "
	82	+ "sensitive values will be treated as plaintext."
	83	+ )
	84	+ return None
	85	+ logger.info("crypto: master key loaded from environment variable.")
	86	+ return parse_master_key_from_string(key_base64)
	87	+
	88	+ @staticmethod
	89	+ def generate_random_salt(length: int = 16) -> bytes:
	90	+ return get_random_bytes(length)
	91	+
	92	+ @staticmethod
	93	+ def hkdf_derive(master_key: bytes, salt: bytes) -> bytes:
	94	+ return HKDF(master_key, 32, salt, SHA256, context=b"database-url-salt")
	95	+
	96	+ def encrypt(self, plaintext: str) -> str:
	97	+ if not plaintext:
	98	+ return plaintext
	99	+ if not self.master_key:
	100	+ return plaintext
	101	+ try:
	102	+ salt = self.generate_random_salt()
	103	+ nonce = get_random_bytes(12)
	104	+ encryption_key = self.hkdf_derive(self.master_key, salt)
	105	+ cipher = AES.new(encryption_key, AES.MODE_GCM, nonce=nonce)
	106	+ ciphertext, auth_tag = cipher.encrypt_and_digest(plaintext.encode("utf-8"))
	107	+ combined_data = salt + nonce + ciphertext + auth_tag
	108	+ return base64.b64encode(combined_data).decode("utf-8")
	109	+ except Exception as e:
	110	+ logger.error("Encryption failed: %s", e)
	111	+ raise ValueError(f"Encryption failed: {e}") from e
	112	+
	113	+ def decrypt(self, ciphertext: str) -> str:
	114	+ if not ciphertext:
	115	+ return ciphertext
	116	+ if not self.master_key:
	117	+ return ciphertext
	118	+ try:
	119	+ data = base64.b64decode(ciphertext)
	120	+ except Exception:
	121	+ logger.info("Base64 decode failed; treating value as plaintext.")
	122	+ return ciphertext
	123	+
	124	+ min_encrypted_len = 16 + 12 + 16
	125	+ if len(data) < min_encrypted_len:
	126	+ return ciphertext
	127	+
	128	+ try:
	129	+ salt_len = 16
	130	+ nonce_len = 12
	131	+ tag_len = 16
	132	+ salt = data[:salt_len]
	133	+ nonce = data[salt_len:salt_len + nonce_len]
	134	+ ciphertext_bytes = data[salt_len + nonce_len:-tag_len]
	135	+ auth_tag = data[-tag_len:]
	136	+ encryption_key = self.hkdf_derive(self.master_key, salt)
	137	+ cipher = AES.new(encryption_key, AES.MODE_GCM, nonce=nonce)
	138	+ plaintext = cipher.decrypt_and_verify(ciphertext_bytes, auth_tag)
	139	+ return plaintext.decode("utf-8")
	140	+ except Exception as e:
	141	+ logger.debug("Decryption failed; treating value as plaintext: %s", e)
	142	+ return ciphertext
	143	+
	144	+ @staticmethod
	145	+ def generate_key() -> str:
	146	+ return base64.b64encode(get_random_bytes(32)).decode("utf-8")
	147	+
	148	+
	149	+_crypto_utils: Optional[CryptoUtils] = None
	150	+
	151	+
	152	+def get_crypto_utils(master_key: Optional[bytes] = None) -> CryptoUtils:
	153	+ global _crypto_utils
	154	+ if _crypto_utils is None:
	155	+ _crypto_utils = CryptoUtils(master_key)
	156	+ return _crypto_utils
	157	+
	158	+
	159	+def reset_crypto_utils_singleton() -> None:
	160	+ """测试或热重载场景下重置单例。"""
	161	+ global _crypto_utils
	162	+ _crypto_utils = None
	163	+
	164	+
	165	+def decrypt_config_value(value: Optional[str]) -> Optional[str]:
	166	+ """
	167	+ 对可能已加密的配置字符串解密；无主密钥或解密失败时返回原值（明文兼容）。
	168	+ """
	169	+ if value is None or value == "":
	170	+ return value
	171	+ return get_crypto_utils().decrypt(value)

Aapplications/a2a_service/common/events.py+348-0

@@ -0,0 +1,348 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+"""
	5	+EDPAgent 事件协议（与 A2A SDK 完全解耦）。
	6	+
	7	+本模块对齐《动态规划 Agent 综合需求文档》§4.5 的完整事件序列：
	8	+ 会话：conversation_start / conversation_end
	9	+ 思考：think_start / think_chunk / think_end
	10	+ 规划：todolist_start / todolist_item / todolist_end
	11	+ 任务：todo_start / todo_status / todo_end
	12	+ 工具：tool_start / tool_status / tool_end
	13	+ 中断：interrupt_start / interrupt_end
	14	+ 总结：final_answer_start / final_answer_chunk / final_answer_end
	15	+
	16	+另外保留：
	17	+ ThoughtEvent / AnswerEvent —— 原版兼容事件（agent_adapter 中仍有映射）
	18	+ DelegateRequest —— Executor 专用（VA 委托路径）
	19	+
	20	+所有事件只依赖 pydantic，不引用 a2a.* 模块。
	21	+"""
	22	+from __future__ import annotations
	23	+
	24	+from typing import Any, Literal, Optional, Union
	25	+
	26	+from pydantic import BaseModel, Field
	27	+
	28	+
	29	+# ════════════════════════════════════════════════════════════════════
	30	+# 会话事件
	31	+# ════════════════════════════════════════════════════════════════════
	32	+
	33	+
	34	+class ConversationStartEvent(BaseModel):
	35	+ """对话开启。每次北向请求开始时发出一次。"""
	36	+ type: Literal["conversation_start"] = "conversation_start"
	37	+ content: str = ""
	38	+
	39	+
	40	+class ConversationEndEvent(BaseModel):
	41	+ """对话结束。整个处理流程结束时发出一次。"""
	42	+ type: Literal["conversation_end"] = "conversation_end"
	43	+ content: str = ""
	44	+
	45	+
	46	+# ════════════════════════════════════════════════════════════════════
	47	+# 思考事件（从 llm_reasoning 流中分段）
	48	+# ════════════════════════════════════════════════════════════════════
	49	+
	50	+
	51	+class ThinkStartEvent(BaseModel):
	52	+ """LLM 思考开始（每轮 ReAct 的第一个 reasoning chunk）。"""
	53	+ type: Literal["think_start"] = "think_start"
	54	+ content: str = ""
	55	+
	56	+
	57	+class ThinkChunkEvent(BaseModel):
	58	+ """LLM 思考流式片段。"""
	59	+ type: Literal["think_chunk"] = "think_chunk"
	60	+ content: str
	61	+
	62	+
	63	+class ThinkEndEvent(BaseModel):
	64	+ """LLM 思考结束（reasoning 流结束或切换为 tool_call / answer）。"""
	65	+ type: Literal["think_end"] = "think_end"
	66	+ content: str = ""
	67	+
	68	+
	69	+# ════════════════════════════════════════════════════════════════════
	70	+# 规划事件（Todolist，由 LLM 在 thought 中按约定 JSON 输出）
	71	+# ════════════════════════════════════════════════════════════════════
	72	+
	73	+
	74	+class TodoListStartEvent(BaseModel):
	75	+ """Todolist 生成开始。"""
	76	+ type: Literal["todolist_start"] = "todolist_start"
	77	+ content: str = ""
	78	+
	79	+
	80	+class TodoListItemEvent(BaseModel):
	81	+ """Todolist 中的单个任务条目。"""
	82	+ type: Literal["todolist_item"] = "todolist_item"
	83	+ id: int \| str
	84	+ title: str
	85	+ status: Literal["pending", "in_progress", "done", "failed"] = "pending"
	86	+ content: str = ""
	87	+
	88	+
	89	+class TodoListEndEvent(BaseModel):
	90	+ """Todolist 生成完成。"""
	91	+ type: Literal["todolist_end"] = "todolist_end"
	92	+ count: int = 0
	93	+ content: str = ""
	94	+
	95	+
	96	+# ════════════════════════════════════════════════════════════════════
	97	+# 任务事件（单个 todo 状态变更，由 LLM 在 thought 中按约定 JSON 输出）
	98	+# ════════════════════════════════════════════════════════════════════
	99	+
	100	+
	101	+class TodoStartEvent(BaseModel):
	102	+ """单个 Todo 开始执行。"""
	103	+ type: Literal["todo_start"] = "todo_start"
	104	+ id: int \| str
	105	+ title: str = ""
	106	+ content: str = ""
	107	+
	108	+
	109	+class TodoStatusEvent(BaseModel):
	110	+ """单个 Todo 状态变更（执行中）。"""
	111	+ type: Literal["todo_status"] = "todo_status"
	112	+ id: int \| str
	113	+ status: Literal["pending", "in_progress", "done", "failed"]
	114	+ content: str = ""
	115	+
	116	+
	117	+class TodoEndEvent(BaseModel):
	118	+ """单个 Todo 执行结束。"""
	119	+ type: Literal["todo_end"] = "todo_end"
	120	+ id: int \| str
	121	+ status: Literal["done", "failed"] = "done"
	122	+ content: str = ""
	123	+
	124	+
	125	+# ════════════════════════════════════════════════════════════════════
	126	+# 工具事件（从 Runner 的 tool_start / tool_end 映射）
	127	+# ════════════════════════════════════════════════════════════════════
	128	+
	129	+
	130	+class ToolStartEvent(BaseModel):
	131	+ """工具调用开始。"""
	132	+ type: Literal["tool_start"] = "tool_start"
	133	+ content: str = ""
	134	+ plugin: str = Field(default="", description="工具名")
	135	+ args: dict[str, Any] = Field(default_factory=dict, description="工具入参")
	136	+
	137	+
	138	+class ToolStatusEvent(BaseModel):
	139	+ """工具调用进行中（可选，用于长时任务的进度）。"""
	140	+ type: Literal["tool_status"] = "tool_status"
	141	+ plugin: str = ""
	142	+ content: str = ""
	143	+ progress: Optional[float] = None # 0.0 - 1.0
	144	+
	145	+
	146	+class ToolEndEvent(BaseModel):
	147	+ """工具调用结束。"""
	148	+ type: Literal["tool_end"] = "tool_end"
	149	+ content: str = ""
	150	+ plugin: str = Field(default="", description="工具名")
	151	+ data: dict[str, Any] = Field(default_factory=dict, description="工具返回数据")
	152	+
	153	+
	154	+# ════════════════════════════════════════════════════════════════════
	155	+# 执行轨迹事件（由 Executor 在 step 边界发射）
	156	+# ════════════════════════════════════════════════════════════════════
	157	+
	158	+
	159	+class PlanningExecutionProcessEvent(BaseModel):
	160	+ """执行轨迹（step 边界标记）。
	161	+
	162	+ 形如 "[执行轨迹] 正在执行步骤N: <desc> (tool=<tool_name>)" 的字符串在 content
	163	+ 里提供给前端做"当前卡在哪一步"的展示。
	164	+
	165	+ 此事件在北向包装里独有 error_code: "" 字段（对齐抓包）。
	166	+ """
	167	+ type: Literal["planning_execution_process"] = "planning_execution_process"
	168	+ content: str = ""
	169	+
	170	+
	171	+# ════════════════════════════════════════════════════════════════════
	172	+# 中断事件（HITL）
	173	+# ════════════════════════════════════════════════════════════════════
	174	+
	175	+
	176	+class InterruptStartEvent(BaseModel):
	177	+ """等待用户输入。"""
	178	+ type: Literal["interrupt_start"] = "interrupt_start"
	179	+ interrupt_id: str
	180	+ content: str = Field(default="", description="提示话术")
	181	+ context: dict[str, Any] = Field(default_factory=dict)
	182	+
	183	+
	184	+class InterruptEndEvent(BaseModel):
	185	+ """中断已恢复（用户输入被接受）。"""
	186	+ type: Literal["interrupt_end"] = "interrupt_end"
	187	+ interrupt_id: str = ""
	188	+ content: str = Field(default="", description="回显确认信息")
	189	+
	190	+
	191	+# ════════════════════════════════════════════════════════════════════
	192	+# 总结事件（从 llm_output / answer 流分段）
	193	+# ════════════════════════════════════════════════════════════════════
	194	+
	195	+
	196	+class FinalAnswerStartEvent(BaseModel):
	197	+ """最终回答开始。"""
	198	+ type: Literal["final_answer_start"] = "final_answer_start"
	199	+ content: str = ""
	200	+
	201	+
	202	+class SummaryEvent(BaseModel):
	203	+ """流式总结片段（按 LLM token 切片逐段发出）。
	204	+
	205	+ 规范定义（feat-north-api-sse.md §4.5.9 / north-api-response-format.md §4.9）：
	206	+ - `summary`：流式通道，承担"边想边说"的 UI 展示
	207	+ - `final_answer_chunk`：一次性全量通道，承担"最终权威文本"
	208	+ 两者并存，互补不冲突。前端流式展示用 SummaryEvent，存档/回显用
	209	+ FinalAnswerChunkEvent。
	210	+ """
	211	+ type: Literal["summary"] = "summary"
	212	+ content: str
	213	+
	214	+
	215	+class FinalAnswerChunkEvent(BaseModel):
	216	+ """最终回答的一次性全量文本帧。
	217	+
	218	+ 注意：抓包与规范一致 —— 这一帧并非流式片段，而是在所有 `summary` 流式
	219	+ 片段结束后，以单帧形式发送完整版（供校对/回显/存档使用）。
	220	+
	221	+ 流式片段请使用 SummaryEvent。
	222	+ """
	223	+ type: Literal["final_answer_chunk"] = "final_answer_chunk"
	224	+ content: str
	225	+
	226	+
	227	+class FinalAnswerEndEvent(BaseModel):
	228	+ """最终回答结束。"""
	229	+ type: Literal["final_answer_end"] = "final_answer_end"
	230	+ content: str = ""
	231	+
	232	+
	233	+# ════════════════════════════════════════════════════════════════════
	234	+# Executor 专用（非北向事件，不出流）
	235	+# ════════════════════════════════════════════════════════════════════
	236	+
	237	+
	238	+class DelegateRequest(BaseModel):
	239	+ """
	240	+ Agent 需要外部 Agent 处理子任务时 yield 的委托对象。
	241	+
	242	+ 协议约定：Orchestrator 收到此对象后：
	243	+ 1. 调用 target_agent 完成子任务
	244	+ 2. 获得 workflow_result（dict）
	245	+ 3. 以 cascade_result=workflow_result 再次调用 agent_stream()
	246	+ """
	247	+ type: Literal["delegate"] = "delegate"
	248	+ intent: str = Field(description="意图标识")
	249	+ target_agent: str \| None = None
	250	+ task_description: str = Field(description="自然语言任务描述")
	251	+
	252	+
	253	+# ════════════════════════════════════════════════════════════════════
	254	+# 兼容事件（保留旧版 API，逐步废弃）
	255	+# ════════════════════════════════════════════════════════════════════
	256	+
	257	+
	258	+class ThoughtEvent(BaseModel):
	259	+ """[兼容] 旧版合并的 thought 事件，建议改用 Think* 系列。"""
	260	+ type: Literal["thought"] = "thought"
	261	+ content: str
	262	+
	263	+
	264	+class AnswerEvent(BaseModel):
	265	+ """[兼容] 旧版合并的 answer 事件，建议改用 FinalAnswer* 系列。"""
	266	+ type: Literal["answer"] = "answer"
	267	+ content: str
	268	+ final: bool = False
	269	+
	270	+
	271	+# ════════════════════════════════════════════════════════════════════
	272	+# 统一 Union 类型
	273	+# ════════════════════════════════════════════════════════════════════
	274	+
	275	+
	276	+AgentEvent = Union[
	277	+ # 会话
	278	+ ConversationStartEvent,
	279	+ ConversationEndEvent,
	280	+ # 思考
	281	+ ThinkStartEvent,
	282	+ ThinkChunkEvent,
	283	+ ThinkEndEvent,
	284	+ # 规划
	285	+ TodoListStartEvent,
	286	+ TodoListItemEvent,
	287	+ TodoListEndEvent,
	288	+ # 任务
	289	+ TodoStartEvent,
	290	+ TodoStatusEvent,
	291	+ TodoEndEvent,
	292	+ # 工具
	293	+ ToolStartEvent,
	294	+ ToolStatusEvent,
	295	+ ToolEndEvent,
	296	+ # 执行轨迹
	297	+ PlanningExecutionProcessEvent,
	298	+ # 中断
	299	+ InterruptStartEvent,
	300	+ InterruptEndEvent,
	301	+ # 总结
	302	+ FinalAnswerStartEvent,
	303	+ SummaryEvent,
	304	+ FinalAnswerChunkEvent,
	305	+ FinalAnswerEndEvent,
	306	+ # Executor
	307	+ DelegateRequest,
	308	+ # 兼容
	309	+ ThoughtEvent,
	310	+ AnswerEvent,
	311	+]
	312	+
	313	+
	314	+# 事件 type → 事件类（给 agent_adapter / user_router 用）
	315	+EVENT_TYPE_MAP = {
	316	+ # 会话
	317	+ "conversation_start": ConversationStartEvent,
	318	+ "conversation_end": ConversationEndEvent,
	319	+ # 思考
	320	+ "think_start": ThinkStartEvent,
	321	+ "think_chunk": ThinkChunkEvent,
	322	+ "think_end": ThinkEndEvent,
	323	+ # 规划
	324	+ "todolist_start": TodoListStartEvent,
	325	+ "todolist_item": TodoListItemEvent,
	326	+ "todolist_end": TodoListEndEvent,
	327	+ # 任务
	328	+ "todo_start": TodoStartEvent,
	329	+ "todo_status": TodoStatusEvent,
	330	+ "todo_end": TodoEndEvent,
	331	+ # 工具
	332	+ "tool_start": ToolStartEvent,
	333	+ "tool_status": ToolStatusEvent,
	334	+ "tool_end": ToolEndEvent,
	335	+ # 执行轨迹
	336	+ "planning_execution_process": PlanningExecutionProcessEvent,
	337	+ # 中断
	338	+ "interrupt_start": InterruptStartEvent,
	339	+ "interrupt_end": InterruptEndEvent,
	340	+ # 总结
	341	+ "final_answer_start": FinalAnswerStartEvent,
	342	+ "summary": SummaryEvent,
	343	+ "final_answer_chunk": FinalAnswerChunkEvent,
	344	+ "final_answer_end": FinalAnswerEndEvent,
	345	+ # 兼容
	346	+ "thought": ThoughtEvent,
	347	+ "answer": AnswerEvent,
	348	+}

Aapplications/a2a_service/common/logger.py+285-0

@@ -0,0 +1,285 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+"""
	5	+日志模板函数。
	6	+"""
	7	+import contextvars
	8	+import json
	9	+import logging
	10	+from contextlib import contextmanager
	11	+from dataclasses import dataclass
	12	+from datetime import datetime, timezone
	13	+from enum import StrEnum
	14	+from typing import Any, Optional
	15	+
	16	+from fastapi import Request
	17	+from loguru import logger
	18	+from pydantic import BaseModel
	19	+
	20	+_TRACE_ID_CTX: contextvars.ContextVar[str] = contextvars.ContextVar(
	21	+ "applications_logger_trace_id", default="unknown"
	22	+)
	23	+_AGENT_ID_CTX: contextvars.ContextVar[str] = contextvars.ContextVar(
	24	+ "applications_logger_agent_id", default="unknown"
	25	+)
	26	+_CONVERSATION_ID_CTX: contextvars.ContextVar[str] = contextvars.ContextVar(
	27	+ "applications_logger_conversation_id", default="unknown"
	28	+)
	29	+
	30	+
	31	+class TagTrace(BaseModel):
	32	+ id: Optional[str] = None # trace_id
	33	+ timestamp: str # 接口开始的时间戳信息
	34	+ name: Optional[str] = None # 可以自定义, 可以是接口Path
	35	+ user_id: Optional[str] = None # 接口传入的用户ID信息
	36	+ session_id: Optional[str] = None # 接口传入的conversation_id
	37	+ input: Optional[dict] = None # 接口调用信息，可以包含body分别按照Json的字段写入
	38	+ output: Optional[dict] = None # 接口返回信息，已有的信息都可以记录到Json中
	39	+ metadata: Optional[dict] = None # 附加信息记录
	40	+ tags: Optional[list] = None # 时间TAG可以记录这里,
	41	+ release: Optional[str] = None # Agent版本信息
	42	+
	43	+
	44	+class ObservationType(StrEnum):
	45	+ SPAN = "SPAN" # 通用操作 / 跨度
	46	+ EVENT = "EVENT" # 时间点事件
	47	+ GENERATION = "GENERATION" # LLM生成操作
	48	+ AGENT = "AGENT" # 自主代理操作
	49	+ TOOL = "TOOL" # 工具调用
	50	+ CHAIN = "CHAIN" # 链式操作
	51	+ RETRIEVER = "RETRIEVER" # 文档检索
	52	+ EVALUATOR = "EVALUATOR" # 质量评估
	53	+ EMBEDDING = "EMBEDDING" # 向量生成
	54	+ GUARDRAIL = "GUARDRAIL" # 安全检查
	55	+
	56	+
	57	+class TagObservation(BaseModel):
	58	+ # 一个阶段的唯一id,暂时使用时间戳来代替id，保证start和end的id是一样的
	59	+ id: str
	60	+ # 暂时可以先不传，工具从日志头解析
	61	+ trace_id: Optional[str] = None
	62	+ # 父Observation的ID，如果是接口调用下的感知、规划、执行、反思阶段，没有父ID，
	63	+ # 如果是模型调用、执行todolist、调用Versatile等属于规划、执行等阶段的更细粒度步骤，
	64	+ # 父ID为其上一层阶段的ID,现阶段置空
	65	+ parent_observation_id: Optional[str] = None
	66	+ # 可根据阶段类型匹配，例如某一跨度的阶段可以使用SPAN，模型调用使用GENERATION，工具调用使用TOOL等
	67	+ type: Optional[ObservationType] = None
	68	+ name: Optional[str] = None
	69	+ start_time: Optional[str] = None
	70	+ end_time: Optional[str] = None
	71	+ input: Optional[dict] = None
	72	+ output: Optional[dict] = None
	73	+ metadata: Optional[dict] = None
	74	+ status_message: Optional[int] = None
	75	+ model: Optional[str] = None
	76	+ internal_model_id: Optional[str] = None # 模型调用阶段的模型ID, 现阶段不填
	77	+ model_parameters: Optional[dict] = None # 模型调用阶段的模型参数信息, 现阶段不填
	78	+ usage_details: Optional[dict[str, int]] = None # 模型调用阶段的模型使用信息，例如token数等, 现阶段不填,
	79	+ cost_details: Optional[dict[str, int]] = None # 模型调用阶段的模型耗时信息，例如Cost数等, 暂时只记录首token耗时,
	80	+ total_cost: Optional[int] = None # 模型调用阶段的调用总耗时, 或者接口 / 工具调用的耗时,
	81	+ completion_start_time: Optional[str] = None # 模型调用阶段的流式响应开始时间, 现阶段不填,
	82	+ prompt_id: Optional[str] = None # 模型调用阶段使用的Prompt ID, 现阶段不填
	83	+
	84	+
	85	+class Tag(StrEnum):
	86	+ TAG_HTTP_REQUEST_START = "HTTP_REQUEST_START"
	87	+ TAG_HTTP_REQUEST_END = "HTTP_REQUEST_END"
	88	+ TAG_AGENT_INIT_TOOLLIST = "AGENT_INIT_TOOLLIST"
	89	+ TAG_LLM_CALL_START = "LLM_CALL_START"
	90	+ TAG_LLM_CALL_END = "LLM_CALL_END"
	91	+ TAG_PLANNING_DECISION = "PLANNING_DECISION"
	92	+ TAG_TODOLIST_QUERY = "TODOLIST_QUERY"
	93	+ TAG_TODOLIST_SAVE = "TODOLIST_SAVE"
	94	+ TAG_SKILL_EXECUTE_START = "SKILL_EXECUTE_START"
	95	+ TAG_SKILL_EXECUTE_END = "SKILL_EXECUTE_END"
	96	+ TAG_VERSATILE_START = "VERSATILE_START"
	97	+ TAG_VERSATILE_END = "VERSATILE_END"
	98	+
	99	+
	100	+class Extra(BaseModel):
	101	+ tag: Optional[Tag] = None
	102	+ cost: Optional[int] = None
	103	+ source: Optional[str] = None
	104	+ user: Optional[str] = None
	105	+ result: Optional[str] = None
	106	+ terminal: Optional[str] = None
	107	+
	108	+
	109	+@dataclass(frozen=True)
	110	+class LogContext:
	111	+ trace_id: str
	112	+ agent_id: str
	113	+ conversation_id: str
	114	+
	115	+
	116	+@dataclass(frozen=True)
	117	+class HttpRequestTagContext:
	118	+ log_context: LogContext
	119	+ request_path: str
	120	+ content_type: str
	121	+ request_headers: dict[str, str]
	122	+ request_body_snapshot: Any
	123	+ user_id: str
	124	+
	125	+
	126	+def to_logger(level: int \| str = logging.INFO, message: Any = "", extra: Extra \| None = None):
	127	+ if extra is not None:
	128	+ if isinstance(message, BaseModel):
	129	+ message = message.model_dump_json(exclude_none=True)
	130	+ with logger.contextualize(**extra.model_dump(exclude_none=True)):
	131	+ logger.log(level, message)
	132	+ else:
	133	+ logger.log(level, message)
	134	+
	135	+
	136	+def current_local_time() -> str:
	137	+ # 生成毫秒精度本地时间字符串，显式指定 UTC 时区后转换为本地时区以避免时区歧义
	138	+ return (
	139	+ datetime.now(tz=timezone.utc)
	140	+ .astimezone()
	141	+ .strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
	142	+ )
	143	+
	144	+
	145	+def extract_header_value(headers: dict[str, str], name: str) -> str:
	146	+ target = name.lower()
	147	+ for key, value in headers.items():
	148	+ if key.lower() == target:
	149	+ return value
	150	+ return ""
	151	+
	152	+
	153	+def current_tag_context() -> tuple[str, str, str]:
	154	+ return _TRACE_ID_CTX.get(), _AGENT_ID_CTX.get(), _CONVERSATION_ID_CTX.get()
	155	+
	156	+
	157	+def build_log_context(trace_id: str, agent_id: str, conversation_id: str) -> LogContext:
	158	+ return LogContext(
	159	+ trace_id=trace_id or "unknown",
	160	+ agent_id=agent_id or "unknown",
	161	+ conversation_id=conversation_id or "unknown",
	162	+ )
	163	+
	164	+
	165	+@contextmanager
	166	+def bind_context(log_context: LogContext):
	167	+ trace_token = _TRACE_ID_CTX.set(log_context.trace_id)
	168	+ agent_token = _AGENT_ID_CTX.set(log_context.agent_id)
	169	+ conversation_token = _CONVERSATION_ID_CTX.set(log_context.conversation_id)
	170	+ try:
	171	+ with logger.contextualize(
	172	+ trace_id=log_context.trace_id,
	173	+ agent_id=log_context.agent_id,
	174	+ conversation_id=log_context.conversation_id,
	175	+ ):
	176	+ yield
	177	+ finally:
	178	+ _TRACE_ID_CTX.reset(trace_token)
	179	+ _AGENT_ID_CTX.reset(agent_token)
	180	+ _CONVERSATION_ID_CTX.reset(conversation_token)
	181	+
	182	+
	183	+async def build_http_request_tag_context(
	184	+ *,
	185	+ request: Request,
	186	+ trace_id: str,
	187	+ agent_id: str,
	188	+ conversation_id: str,
	189	+) -> HttpRequestTagContext:
	190	+ request_headers = dict(request.headers)
	191	+ content_type = request.headers.get("content-type", "").lower()
	192	+ request_body_snapshot: Any = {"raw_body": ""}
	193	+
	194	+ try:
	195	+ raw_body = await request.body()
	196	+ raw_body_text = raw_body.decode("utf-8", errors="replace")
	197	+ if raw_body_text:
	198	+ if "application/json" in content_type:
	199	+ try:
	200	+ request_body_snapshot = json.loads(raw_body_text)
	201	+ except Exception:
	202	+ request_body_snapshot = {"raw_body": raw_body_text}
	203	+ else:
	204	+ request_body_snapshot = {"raw_body": raw_body_text}
	205	+ except Exception:
	206	+ request_body_snapshot = {"raw_body": "<unavailable>"}
	207	+
	208	+ user_id = extract_header_value(
	209	+ request_headers, "x-user-id"
	210	+ ) or extract_header_value(request_headers, "cust-userid")
	211	+
	212	+ return HttpRequestTagContext(
	213	+ log_context=build_log_context(trace_id, agent_id, conversation_id),
	214	+ request_path=request.url.path,
	215	+ content_type=content_type,
	216	+ request_headers=request_headers,
	217	+ request_body_snapshot=request_body_snapshot,
	218	+ user_id=user_id,
	219	+ )
	220	+
	221	+
	222	+def build_http_trace(
	223	+ *,
	224	+ http_request_tag_context: HttpRequestTagContext,
	225	+ input_payload: Optional[dict[str, Any]] = None,
	226	+ output_payload: Optional[dict[str, Any]] = None,
	227	+ metadata: Optional[dict[str, Any]] = None,
	228	+ tags: Optional[list[str]] = None,
	229	+ release: str = "1.0.0",
	230	+) -> TagTrace:
	231	+ trace_id = http_request_tag_context.log_context.trace_id
	232	+ return TagTrace(
	233	+ id=trace_id,
	234	+ timestamp=current_local_time(),
	235	+ name=http_request_tag_context.request_path,
	236	+ user_id=http_request_tag_context.user_id,
	237	+ session_id=http_request_tag_context.log_context.conversation_id,
	238	+ input=input_payload if input_payload is not None else {},
	239	+ output=output_payload if output_payload is not None else {},
	240	+ metadata=metadata if metadata is not None else {"UNION_NO": trace_id},
	241	+ tags=tags if tags is not None else [],
	242	+ release=release,
	243	+ )
	244	+
	245	+
	246	+def build_versatile_start_observation(
	247	+ *,
	248	+ call_id: str,
	249	+ name: str,
	250	+ request_headers: dict[str, Any],
	251	+ request_body: Any,
	252	+) -> TagObservation:
	253	+ trace_id, _, _ = current_tag_context()
	254	+ return TagObservation(
	255	+ id=call_id,
	256	+ trace_id=trace_id,
	257	+ type=ObservationType.TOOL,
	258	+ name=name,
	259	+ start_time=current_local_time(),
	260	+ input={
	261	+ "request_header": request_headers,
	262	+ "request_body": request_body,
	263	+ },
	264	+ )
	265	+
	266	+
	267	+def build_versatile_end_observation(
	268	+ *,
	269	+ call_id: str,
	270	+ name: str,
	271	+ output_payload: dict[str, Any],
	272	+ status_message: Any,
	273	+ duration_ms: int,
	274	+) -> TagObservation:
	275	+ trace_id, _, _ = current_tag_context()
	276	+ return TagObservation(
	277	+ id=call_id,
	278	+ trace_id=trace_id,
	279	+ type=ObservationType.TOOL,
	280	+ name=name,
	281	+ end_time=current_local_time(),
	282	+ output=output_payload,
	283	+ status_message=status_message,
	284	+ total_cost=max(duration_ms, 0),
	285	+ )

Aapplications/a2a_service/common/redis_client.py+99-0

@@ -0,0 +1,99 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+"""
	5	+异步 Redis 客户端（redis-py asyncio）。
	6	+"""
	7	+from __future__ import annotations
	8	+
	9	+import json
	10	+from typing import Any, Optional
	11	+from urllib.parse import urlsplit, urlunsplit
	12	+
	13	+from loguru import logger
	14	+from redis.asyncio import Redis
	15	+
	16	+
	17	+_RELEASE_LOCK_LUA = """
	18	+if redis.call('GET', KEYS[1]) == ARGV[1] then
	19	+ return redis.call('DEL', KEYS[1])
	20	+end
	21	+return 0
	22	+"""
	23	+
	24	+
	25	+class RedisClient:
	26	+ """轻量 Redis 封装，仅暴露项目用到的操作。"""
	27	+
	28	+ def __init__(self) -> None:
	29	+ self._client: Optional[Redis] = None
	30	+
	31	+ async def connect(self, url: str) -> None:
	32	+ self._client = Redis.from_url(url, decode_responses=True, protocol=2)
	33	+ await self._client.ping()
	34	+ safe_url = url
	35	+ parsed = urlsplit(url)
	36	+ if parsed.password is not None:
	37	+ username = parsed.username or ""
	38	+ host = parsed.hostname or ""
	39	+ if host and ":" in host and not host.startswith("["):
	40	+ host = f"[{host}]"
	41	+ port = f":{parsed.port}" if parsed.port is not None else ""
	42	+ masked_netloc = f"{username}:@{host}{port}"
	43	+ safe_url = urlunsplit((parsed.scheme, masked_netloc, parsed.path, parsed.query, parsed.fragment))
	44	+ logger.info(f"[Redis] 已连接：{safe_url}")
	45	+
	46	+ async def disconnect(self) -> None:
	47	+ if self._client:
	48	+ await self._client.aclose()
	49	+ self._client = None
	50	+ logger.info("[Redis] 连接已关闭")
	51	+
	52	+ @property
	53	+ def client(self) -> Redis:
	54	+ if self._client is None:
	55	+ raise RuntimeError("RedisClient 未连接，请先调用 connect()")
	56	+ return self._client
	57	+
	58	+ async def get(self, key: str) -> Optional[str]:
	59	+ return await self.client.get(key)
	60	+
	61	+ async def set(self, key: str, value: str, ex: Optional[int] = None) -> None:
	62	+ await self.client.set(key, value, ex=ex)
	63	+
	64	+ async def set_if_not_exists(self, key: str, value: str, ex: int) -> bool:
	65	+ result = await self.client.set(key, value, ex=ex, nx=True)
	66	+ return bool(result)
	67	+
	68	+ async def set_nx(self, key: str, value: str, ex: Optional[int] = None) -> bool:
	69	+ """兼容旧调用方式：SET if Not eXists。"""
	70	+ if ex is None:
	71	+ result = await self.client.set(key, value, nx=True)
	72	+ return result is not None
	73	+ result = await self.client.set(key, value, ex=ex, nx=True)
	74	+ return result is not None
	75	+
	76	+ async def delete(self, *keys: str) -> None:
	77	+ if keys:
	78	+ await self.client.delete(*keys)
	79	+
	80	+ async def get_json(self, key: str) -> Optional[Any]:
	81	+ raw = await self.get(key)
	82	+ if raw is None:
	83	+ return None
	84	+ try:
	85	+ return json.loads(raw)
	86	+ except json.JSONDecodeError:
	87	+ logger.warning(f"[Redis] JSON 解析失败：key={key}")
	88	+ return None
	89	+
	90	+ async def set_json(self, key: str, value: Any, ex: Optional[int] = None) -> None:
	91	+ await self.set(key, json.dumps(value, ensure_ascii=False), ex=ex)
	92	+
	93	+ async def acquire_lock(self, lock_key: str, owner_id: str, ttl_seconds: int) -> bool:
	94	+ ttl = max(int(ttl_seconds), 1)
	95	+ return await self.set_if_not_exists(lock_key, owner_id, ex=ttl)
	96	+
	97	+ async def release_lock(self, lock_key: str, owner_id: str) -> bool:
	98	+ result = await self.client.eval(_RELEASE_LOCK_LUA, 1, lock_key, owner_id)
	99	+ return int(result or 0) == 1

Aapplications/a2a_service/common/redis_task_store.py+53-0

@@ -0,0 +1,53 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+"""
	5	+Redis-backed A2A TaskStore.
	6	+
	7	+Task protobuf 序列化为 base64 二进制存入 Redis，key 格式：a2a:task:{task_id}。
	8	+TTL 复用 redis_session_ttl 配置（秒），默认 1800 s。
	9	+"""
	10	+from __future__ import annotations
	11	+
	12	+import base64
	13	+from typing import Optional
	14	+
	15	+from a2a.server.context import ServerCallContext
	16	+from a2a.server.tasks.task_store import TaskStore
	17	+from a2a.types.a2a_pb2 import ListTasksRequest, ListTasksResponse, Task
	18	+from loguru import logger
	19	+
	20	+from common.redis_client import RedisClient
	21	+
	22	+_KEY_PREFIX = "a2a:task:"
	23	+
	24	+
	25	+class RedisTaskStore(TaskStore):
	26	+ def __init__(self, redis: RedisClient, ttl: int = 1800) -> None:
	27	+ self._redis = redis
	28	+ self._ttl = ttl
	29	+
	30	+ async def save(self, task: Task, context: ServerCallContext) -> None:
	31	+ key = _KEY_PREFIX + task.id
	32	+ data = base64.b64encode(task.SerializeToString()).decode("ascii")
	33	+ await self._redis.set(key, data, ex=self._ttl)
	34	+ logger.debug(f"[TaskStore] save task={task.id} state={task.status.state}")
	35	+
	36	+ async def get(self, task_id: str, context: ServerCallContext) -> Optional[Task]:
	37	+ if not task_id:
	38	+ return None
	39	+ raw = await self._redis.get(_KEY_PREFIX + task_id)
	40	+ if raw is None:
	41	+ return None
	42	+ task = Task()
	43	+ task.ParseFromString(base64.b64decode(raw))
	44	+ return task
	45	+
	46	+ async def delete(self, task_id: str, context: ServerCallContext) -> None:
	47	+ await self._redis.delete(_KEY_PREFIX + task_id)
	48	+ logger.debug(f"[TaskStore] delete task={task_id}")
	49	+
	50	+ async def list(
	51	+ self, params: ListTasksRequest, context: ServerCallContext
	52	+ ) -> ListTasksResponse:
	53	+ return ListTasksResponse()

Aapplications/a2a_service/common/response_wrapper.py+183-0

@@ -0,0 +1,183 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+"""
	5	+北向响应包装器。
	6	+
	7	+把内部 A2A / EDPAgent 事件包装成前端可见的 JSON 格式，对齐
	8	+docs/feat-north-api-sse.md §4.4.3 与 docs/north-api-response-format.md §3.
	9	+
	10	+两种包装：
	11	+ - agent event：EDPAgent 内部事件（think/todolist/tool/interrupt/summary 等）
	12	+ - workflow event：Versatile 工作流节点转发事件（event=message/end）
	13	+
	14	+还包含错误/限流场景的简化响应（wrap_error）。
	15	+
	16	+字段语义请直接对齐规范文档：
	17	+
	18	+ 1. `execution_time` 对所有 agent event 均为数字（秒），对齐 spec §2.3.3；
	19	+ workflow event 同样为数字。
	20	+ 2. `error_code` 只在 event_type == "planning_execution_process" 时带上空串；
	21	+ 其他 agent event 不带此字段；workflow event 也不带。
	22	+ 3. `output` / `error` 仅 agent event 有，固定空串；workflow event 不带。
	23	+ 4. `latency` 字段保留为空串，为历史预留字段；
	24	+ `plugin` 字段在 tool_* agent event 上承载工具名，其他事件为空串。
	25	+
	26	+本模块不依赖 a2a SDK，只用标准库。
	27	+"""
	28	+from __future__ import annotations
	29	+
	30	+import time
	31	+from typing import Any
	32	+
	33	+# ════════════════════════════════════════════════════════════════════
	34	+# 事件类型白名单
	35	+# ════════════════════════════════════════════════════════════════════
	36	+
	37	+# 这类事件的包装器会额外带一个 `error_code: ""` 字段。
	38	+# 对齐抓包观察：其他事件帧均不带 error_code；限流/错误走 wrap_error。
	39	+_EVENTS_WITH_ERROR_CODE: frozenset[str] = frozenset({
	40	+ "planning_execution_process",
	41	+})
	42	+
	43	+# （历史上 think_chunk 曾被设为 ""，与抓包对齐；现统一以 spec §2.3.3 为准，数字）
	44	+_EVENTS_WITH_EMPTY_EXECUTION_TIME: frozenset[str] = frozenset()
	45	+
	46	+
	47	+# ════════════════════════════════════════════════════════════════════
	48	+# agent event 包装
	49	+# ════════════════════════════════════════════════════════════════════
	50	+
	51	+
	52	+def wrap_agent_event(
	53	+ event_type: str,
	54	+ content: str,
	55	+ data: dict[str, Any] \| None,
	56	+ *,
	57	+ agent_id: str,
	58	+ conversation_id: str,
	59	+ elapsed: float,
	60	+ created_time_ms: int \| None = None,
	61	+ plugin: str = "",
	62	+) -> dict[str, Any]:
	63	+ """包装 EDPAgent 内部事件（think / todolist / tool / interrupt / summary 等）。
	64	+
	65	+ event_type 可以是任意字符串，本函数不做白名单过滤：
	66	+ 业务 skill 发的自定义 type（如 `product_select_progress`）也会被正确包装。
	67	+
	68	+ Args:
	69	+ event_type: 事件类型，对应 custom_rsp_data.event。
	70	+ content: 事件人类可读文本（可含 HTML 片段，如 todolist_item 的 ``<br/>``）。
	71	+ data: 结构化补充，对应 custom_rsp_data.data。None / {} 都可以。
	72	+ agent_id: 当前请求的 agent_id，回显给客户端。
	73	+ conversation_id: 当前请求的 conversation_id。
	74	+ elapsed: 本次 turn 累计耗时秒数（通常 `time.monotonic() - start`）。
	75	+ created_time_ms: 事件产生的 epoch 毫秒；默认取当前时间。
	76	+ plugin: 工具名（tool_* 事件专用），填入 custom_rsp_data.plugin；默认空串。
	77	+
	78	+ Returns:
	79	+ 可直接 json.dumps 的 dict，格式对齐规范文档 §4.4.3（agent event）。
	80	+ """
	81	+ exec_time: float \| str = (
	82	+ "" if event_type in _EVENTS_WITH_EMPTY_EXECUTION_TIME else elapsed
	83	+ )
	84	+ wrapped: dict[str, Any] = {
	85	+ "success": True,
	86	+ "agent_id": agent_id,
	87	+ "conversation_id": conversation_id,
	88	+ "output": "",
	89	+ "error": "",
	90	+ "execution_time": exec_time,
	91	+ "custom_rsp_data": {
	92	+ "data": data or {},
	93	+ "event": event_type,
	94	+ "content": content,
	95	+ "createdTime": (
	96	+ created_time_ms
	97	+ if created_time_ms is not None
	98	+ else int(time.time() * 1000)
	99	+ ),
	100	+ "latency": "",
	101	+ "plugin": plugin,
	102	+ },
	103	+ }
	104	+ if event_type in _EVENTS_WITH_ERROR_CODE:
	105	+ wrapped["error_code"] = ""
	106	+ return wrapped
	107	+
	108	+
	109	+# ════════════════════════════════════════════════════════════════════
	110	+# workflow event 包装
	111	+# ════════════════════════════════════════════════════════════════════
	112	+
	113	+
	114	+def wrap_workflow_event(
	115	+ event_kind: str,
	116	+ data: dict[str, Any],
	117	+ *,
	118	+ agent_id: str,
	119	+ conversation_id: str,
	120	+ elapsed: float,
	121	+) -> dict[str, Any]:
	122	+ """包装从 Versatile 转发的工作流节点事件。
	123	+
	124	+ workflow event 比 agent event 字段更少：无 output / error / error_code 字段，
	125	+ custom_rsp_data 只含 event 和 data 两个键。
	126	+
	127	+ Args:
	128	+ event_kind: "message" 或 "end"。
	129	+ data: Versatile 节点数据（含 text / node_id / node_type / node_name /
	130	+ workflow_id / summary? / is_finished? 等字段）。event_kind=="end"
	131	+ 时通常为空 dict。
	132	+ agent_id / conversation_id: 同 agent event。
	133	+ elapsed: 累计秒数。
	134	+
	135	+ Returns:
	136	+ 可 json.dumps 的 dict，格式对齐规范文档 §3.3（workflow event）。
	137	+ """
	138	+ if event_kind not in ("message", "end"):
	139	+ # 防御：只接受观察到的两个值；其他值也允许透传，避免阻塞将来协议扩展
	140	+ pass
	141	+ return {
	142	+ "success": True,
	143	+ "agent_id": agent_id,
	144	+ "conversation_id": conversation_id,
	145	+ "execution_time": elapsed,
	146	+ "custom_rsp_data": {
	147	+ "event": event_kind,
	148	+ "data": data,
	149	+ },
	150	+ }
	151	+
	152	+
	153	+# ════════════════════════════════════════════════════════════════════
	154	+# 错误 / 限流响应包装
	155	+# ════════════════════════════════════════════════════════════════════
	156	+
	157	+
	158	+def wrap_error(
	159	+ *,
	160	+ agent_id: str,
	161	+ conversation_id: str,
	162	+ elapsed: float,
	163	+ error_code: str,
	164	+ error_msg: str,
	165	+) -> dict[str, Any]:
	166	+ """限流、内部错误等场景的简化响应。
	167	+
	168	+ 与成功帧相比：``success`` 为 False，带 ``error_code`` 和 ``error_msg``，
	169	+ 没有 ``custom_rsp_data``。当前限流分支（user_router.py:434-440）已经是
	170	+ 这种结构，保持兼容。
	171	+
	172	+ Args:
	173	+ error_code: 业务错误码（如限流的 "100001"）。
	174	+ error_msg: 错误中文描述。
	175	+ """
	176	+ return {
	177	+ "success": False,
	178	+ "agent_id": agent_id,
	179	+ "conversation_id": conversation_id,
	180	+ "execution_time": elapsed,
	181	+ "error_code": error_code,
	182	+ "error_msg": error_msg,
	183	+ }

Aapplications/a2a_service/config.py+73-0

@@ -0,0 +1,73 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+from __future__ import annotations
	5	+
	6	+from functools import lru_cache
	7	+from pathlib import Path
	8	+from typing import Optional
	9	+from urllib.parse import quote_plus
	10	+
	11	+from pydantic_settings import BaseSettings, SettingsConfigDict
	12	+
	13	+from common.crypto import decrypt_config_value
	14	+
	15	+
	16	+
	17	+class Settings(BaseSettings):
	18	+ model_config = SettingsConfigDict(
	19	+ env_file=Path(__file__).parent / ".env",
	20	+ env_file_encoding="utf-8",
	21	+ extra="ignore",
	22	+ )
	23	+
	24	+ # ── App ─────────────────────────────────────────────────────────────────
	25	+ app_name: Optional[str] = None
	26	+
	27	+ # ── Redis（会话状态）────────────────────────────────────────────────────
	28	+ redis_host: Optional[str] = None
	29	+ redis_port: Optional[int] = None
	30	+ redis_db: Optional[int] = None
	31	+ redis_password: Optional[str] = None
	32	+ redis_session_ttl: Optional[int] = None
	33	+
	34	+ # ── 启动编排（Redis 锁 + 状态协同）────────────────────────────────────
	35	+ bootstrap_coordination_enabled: bool = True
	36	+ bootstrap_lock_name: str = "a2a_global_bootstrap"
	37	+ bootstrap_lock_ttl_sec: int = 180
	38	+ bootstrap_wait_timeout_sec: int = 300
	39	+ bootstrap_poll_interval_sec: float = 1.0
	40	+
	41	+ # ── 入口限流（与 Orchestrator 限流能力对齐）─────────────────────────────
	42	+ rate_limit_max_requests: int = 1
	43	+ rate_limit_window_seconds: int = 10
	44	+ global_rate_limit_max_requests: int = 10
	45	+ global_rate_limit_window_seconds: int = 10
	46	+
	47	+ # ── VersatileAdapter（内部 A2A 服务地址）────────────────────────────────
	48	+ versatile_adapter_url: Optional[str] = None
	49	+ # VA 流中携带工作流最终结果的 QA 节点名称（node_type=="QA" 且 node_name==此值）
	50	+ va_workflow_result_node: Optional[str] = None
	51	+
	52	+ # ── FastAPI ─────────────────────────────────────────────────────────────
	53	+ fastapi_host: Optional[str] = None
	54	+ fastapi_port: Optional[int] = None
	55	+ fastapi_debug: Optional[bool] = None
	56	+ fastapi_workers: Optional[int] = None
	57	+
	58	+ # ── 日志 ────────────────────────────────────────────────────────────────
	59	+ log_level: Optional[str] = None
	60	+ log_dir: Optional[str] = None
	61	+
	62	+ @property
	63	+ def redis_url(self) -> str:
	64	+ if self.redis_password:
	65	+ plain = decrypt_config_value(self.redis_password) or ""
	66	+ pwd = quote_plus(plain)
	67	+ return f"redis://:{pwd}@{self.redis_host}:{self.redis_port}/{self.redis_db}"
	68	+ return f"redis://{self.redis_host}:{self.redis_port}/{self.redis_db}"
	69	+
	70	+
	71	+@lru_cache
	72	+def get_settings() -> Settings:
	73	+ return Settings()

Aapplications/a2a_service/main.py+33-0

@@ -0,0 +1,33 @@
	1	+#!/usr/bin/env python
	2	+# coding: utf-8
	3	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	4	+
	5	+from __future__ import annotations
	6	+
	7	+import uvicorn
	8	+from loguru import logger
	9	+
	10	+from config import get_settings
	11	+
	12	+
	13	+def main() -> None:
	14	+ settings = get_settings()
	15	+
	16	+ logger.info(f"启动 {settings.app_name}...")
	17	+ logger.info(f"监听地址: {settings.fastapi_host}:{settings.fastapi_port}")
	18	+ logger.info(f"调试模式: {settings.fastapi_debug}")
	19	+ logger.info(f"Worker 数量: {settings.fastapi_workers}")
	20	+
	21	+ uvicorn.run(
	22	+ "app:app",
	23	+ host=settings.fastapi_host,
	24	+ port=settings.fastapi_port,
	25	+ workers=settings.fastapi_workers if not settings.fastapi_debug else 1,
	26	+ reload=settings.fastapi_debug,
	27	+ log_level=settings.log_level.lower(),
	28	+ loop="auto",
	29	+ )
	30	+
	31	+
	32	+if __name__ == "__main__":
	33	+ main()

Aapplications/a2a_service/orchestrator/__init__.py+2-0

@@ -0,0 +1,2 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved

Aapplications/a2a_service/orchestrator/agent_adapter.py+127-0

@@ -0,0 +1,127 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+"""
	5	+Agent 事件 → A2A v1.0 事件转换。
	6	+
	7	+映射规则（对齐需求文档 §4.5）：
	8	+ 除 final_answer_end / AnswerEvent(final=True) 触发 COMPLETED 外，
	9	+ 所有事件统一封装为 TaskArtifactUpdateEvent(last_chunk=False)，
	10	+ 在 parts 里携带：
	11	+ - Part(text=<content>) 可读文本（若有）
	12	+ - Part(data=<event fields>) 结构化数据，客户端按 data.type 分派
	13	+
	14	+ DelegateRequest → None（Executor 直接处理）
	15	+"""
	16	+from __future__ import annotations
	17	+
	18	+import uuid
	19	+from typing import Optional
	20	+
	21	+from a2a.types.a2a_pb2 import (
	22	+ Artifact,
	23	+ Message,
	24	+ Part,
	25	+ TaskArtifactUpdateEvent,
	26	+ TaskStatus,
	27	+ TaskStatusUpdateEvent,
	28	+ ROLE_AGENT,
	29	+ TASK_STATE_COMPLETED,
	30	+)
	31	+from google.protobuf.struct_pb2 import Struct, Value
	32	+
	33	+from common.events import (
	34	+ AgentEvent,
	35	+ # 会话
	36	+ ConversationStartEvent, ConversationEndEvent,
	37	+ # 思考
	38	+ ThinkStartEvent, ThinkChunkEvent, ThinkEndEvent,
	39	+ # 规划
	40	+ TodoListStartEvent, TodoListItemEvent, TodoListEndEvent,
	41	+ # 任务
	42	+ TodoStartEvent, TodoStatusEvent, TodoEndEvent,
	43	+ # 工具
	44	+ ToolStartEvent, ToolStatusEvent, ToolEndEvent,
	45	+ # 执行轨迹
	46	+ PlanningExecutionProcessEvent,
	47	+ # 中断
	48	+ InterruptStartEvent, InterruptEndEvent,
	49	+ # 总结
	50	+ FinalAnswerStartEvent, SummaryEvent, FinalAnswerChunkEvent, FinalAnswerEndEvent,
	51	+ # 兼容
	52	+ ThoughtEvent, AnswerEvent,
	53	+)
	54	+
	55	+
	56	+def _build_data_part(data: dict) -> Part:
	57	+ struct = Struct()
	58	+ # Struct 要求所有值必须是 JSON 可序列化的简单类型；pydantic model_dump 已经做过
	59	+ struct.update(data)
	60	+ value = Value()
	61	+ value.struct_value.CopyFrom(struct)
	62	+ part = Part()
	63	+ part.data.CopyFrom(value)
	64	+ return part
	65	+
	66	+
	67	+def _build_artifact(text: str, data: dict) -> Artifact:
	68	+ parts = []
	69	+ if text:
	70	+ parts.append(Part(text=text))
	71	+ parts.append(_build_data_part(data))
	72	+ return Artifact(artifact_id=str(uuid.uuid4()), parts=parts)
	73	+
	74	+
	75	+def _artifact_event(task_id: str, conv_id: str, artifact: Artifact) -> TaskArtifactUpdateEvent:
	76	+ return TaskArtifactUpdateEvent(
	77	+ task_id=task_id,
	78	+ context_id=conv_id,
	79	+ artifact=artifact,
	80	+ last_chunk=False,
	81	+ )
	82	+
	83	+
	84	+def _completed(task_id: str, conv_id: str, content: str) -> TaskStatusUpdateEvent:
	85	+ part = Part(text=content)
	86	+ msg = Message(role=ROLE_AGENT, message_id=str(uuid.uuid4()), parts=[part])
	87	+ return TaskStatusUpdateEvent(
	88	+ task_id=task_id,
	89	+ context_id=conv_id,
	90	+ status=TaskStatus(state=TASK_STATE_COMPLETED, message=msg),
	91	+ )
	92	+
	93	+
	94	+def agent_event_to_a2a(
	95	+ event: AgentEvent,
	96	+ task_id: str,
	97	+ conv_id: str,
	98	+) -> Optional[TaskArtifactUpdateEvent \| TaskStatusUpdateEvent]:
	99	+ """AgentEvent → A2A 事件。DelegateRequest 返回 None 由 Executor 处理。"""
	100	+
	101	+ # ── 终止态：TaskStatusUpdateEvent(COMPLETED) ─────────────────────
	102	+ if isinstance(event, FinalAnswerEndEvent):
	103	+ return _completed(task_id, conv_id, event.content)
	104	+
	105	+ # 兼容旧 AnswerEvent(final=True)
	106	+ if isinstance(event, AnswerEvent) and event.final:
	107	+ return _completed(task_id, conv_id, event.content)
	108	+
	109	+ # ── 其余事件统一 TaskArtifactUpdateEvent ──────────────────────────
	110	+ event_groups = (
	111	+ ConversationStartEvent, ConversationEndEvent,
	112	+ ThinkStartEvent, ThinkChunkEvent, ThinkEndEvent,
	113	+ TodoListStartEvent, TodoListItemEvent, TodoListEndEvent,
	114	+ TodoStartEvent, TodoStatusEvent, TodoEndEvent,
	115	+ ToolStartEvent, ToolStatusEvent, ToolEndEvent,
	116	+ PlanningExecutionProcessEvent,
	117	+ InterruptStartEvent, InterruptEndEvent,
	118	+ FinalAnswerStartEvent, SummaryEvent, FinalAnswerChunkEvent,
	119	+ ThoughtEvent, AnswerEvent,
	120	+ )
	121	+ if isinstance(event, event_groups):
	122	+ data = event.model_dump()
	123	+ text = str(data.get("content", "") or "")
	124	+ return _artifact_event(task_id, conv_id, _build_artifact(text, data))
	125	+
	126	+ # DelegateRequest 等由上层处理
	127	+ return None

Aapplications/a2a_service/orchestrator/executor.py+713-0

@@ -0,0 +1,713 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+"""
	5	+Executor — 核心编排逻辑（a2a-sdk 1.0.0-alpha.1，全量 v1.0 protobuf）。
	6	+
	7	+职责：
	8	+ 1. 实现 AgentExecutor 接口，由 user_router 或 DefaultRequestHandler 调用
	9	+ 2. 首轮：调用 agent_stream()，处理 DelegateRequest / AnswerEvent
	10	+ 3. DelegateRequest：调用 VersatileAdapter（A2A Client），根据返回决定续轮或挂起
	11	+ 4. 续轮：从 context.current_task 读取 Task 状态，通过 Task.metadata 传递 va_task_id
	12	+
	13	+Task 状态流转（存于 RedisTaskStore）：
	14	+ WORKING → [DelegateRequest + VA 无 end node] → INPUT_REQUIRED（metadata.va_task_id 已写入）
	15	+ INPUT_REQUIRED → [下一轮用户输入 + VA 有 end node] → WORKING → cascade → COMPLETED
	16	+"""
	17	+from __future__ import annotations
	18	+
	19	+import time
	20	+import uuid
	21	+from dataclasses import dataclass
	22	+from typing import Optional
	23	+
	24	+from a2a.client import Client
	25	+from a2a.server.agent_execution import AgentExecutor, RequestContext
	26	+from a2a.server.context import ServerCallContext
	27	+from a2a.server.events import EventQueue
	28	+from a2a.types.a2a_pb2 import (
	29	+ Message,
	30	+ Part,
	31	+ SendMessageRequest,
	32	+ Task,
	33	+ TaskArtifactUpdateEvent,
	34	+ TaskStatus,
	35	+ TaskStatusUpdateEvent,
	36	+ ROLE_USER,
	37	+ TASK_STATE_COMPLETED,
	38	+ TASK_STATE_INPUT_REQUIRED,
	39	+ TASK_STATE_WORKING,
	40	+)
	41	+from google.protobuf.json_format import MessageToDict
	42	+from google.protobuf.struct_pb2 import Struct, Value
	43	+from loguru import logger
	44	+
	45	+from agents.EDPAgent import agent_stream
	46	+from common.constants import session_request_key
	47	+from common.events import (
	48	+ DelegateRequest,
	49	+ PlanningExecutionProcessEvent,
	50	+ ToolStartEvent,
	51	+)
	52	+from common.logger import (
	53	+ Extra,
	54	+ Tag,
	55	+ build_versatile_end_observation,
	56	+ build_versatile_start_observation,
	57	+ to_logger,
	58	+)
	59	+from common.redis_client import RedisClient
	60	+from config import get_settings
	61	+from orchestrator.agent_adapter import agent_event_to_a2a
	62	+from common.redis_task_store import RedisTaskStore
	63	+
	64	+_TTL = 1800
	65	+
	66	+
	67	+def _rewrite_recommend_delegate(intent: str, task_description: str) -> tuple[str, str]:
	68	+ """临时兼容旧链路：推荐首跳改写为平台历史上可识别的入口。"""
	69	+ if intent != "理财推荐":
	70	+ return intent, task_description
	71	+
	72	+ normalized_query = (task_description or "").strip()
	73	+ if not normalized_query or normalized_query == "推荐理财产品":
	74	+ normalized_query = "请推荐低风险理财产品"
	75	+
	76	+ return "理财选品购买", normalized_query
	77	+
	78	+
	79	+@dataclass(frozen=True)
	80	+class _TurnContext:
	81	+ """单轮 Executor 编排所需的上下文。
	82	+
	83	+ 将相关性较强的会话/任务/调用句柄打包传递，避免在内部方法间传入
	84	+ 个数较多的散参数（参考 G.FNM.03）。
	85	+ """
	86	+
	87	+ conv_id: str
	88	+ task_id: str
	89	+ call_context: ServerCallContext
	90	+ event_queue: EventQueue
	91	+
	92	+
	93	+@dataclass(frozen=True)
	94	+class _VaRequestPayload:
	95	+ """构造 VersatileAdapter ``SendMessageRequest`` 的载荷集合。
	96	+
	97	+ ``headers`` / ``body`` / ``params`` 在调用层共同决定下游工作流入参，
	98	+ 通过统一的 dataclass 进行命名封装（参考 G.FNM.03）。
	99	+ """
	100	+
	101	+ query: str
	102	+ headers: dict
	103	+ body: dict
	104	+ params: Optional[dict] = None
	105	+ task_id: str = ""
	106	+ conv_id: str = ""
	107	+
	108	+
	109	+class Executor(AgentExecutor):
	110	+ def __init__(
	111	+ self, va_client: Client, redis: RedisClient, task_store: RedisTaskStore
	112	+ ) -> None:
	113	+ self._va_client = va_client
	114	+ self._redis = redis
	115	+ self._task_store = task_store
	116	+
	117	+ async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
	118	+ conv_id = context.context_id or ""
	119	+ task_id = context.task_id or str(uuid.uuid4())
	120	+ call_context = context.call_context
	121	+ current_task = context.current_task
	122	+
	123	+ # 从 message parts 提取 text、body
	124	+ user_query = ""
	125	+ original_headers: dict = {}
	126	+ original_body: dict = {}
	127	+ if context.message:
	128	+ for part in context.message.parts:
	129	+ which = part.WhichOneof("content")
	130	+ if which == "text" and not user_query:
	131	+ user_query = part.text
	132	+ elif which == "data":
	133	+ data = MessageToDict(part.data)
	134	+ if isinstance(data, dict):
	135	+ original_headers = data.get("headers", {})
	136	+ original_body = data.get("body", data)
	137	+
	138	+ turn_ctx = _TurnContext(
	139	+ conv_id=conv_id,
	140	+ task_id=task_id,
	141	+ call_context=call_context,
	142	+ event_queue=event_queue,
	143	+ )
	144	+
	145	+ # ── 续轮路径：Task 处于 INPUT_REQUIRED（VA 上次未完成）───────────────
	146	+ if current_task and current_task.status.state == TASK_STATE_INPUT_REQUIRED:
	147	+ meta = MessageToDict(current_task.metadata)
	148	+ va_task_id = meta.get("va_task_id", "")
	149	+ logger.info(
	150	+ f"[Executor] INPUT_REQUIRED 续轮：conv={conv_id}, va_task={va_task_id}"
	151	+ )
	152	+ await self._continue_versatile_adapter(
	153	+ turn_ctx,
	154	+ va_task_id=va_task_id,
	155	+ user_input=user_query,
	156	+ headers=original_headers,
	157	+ original_body=original_body,
	158	+ )
	159	+ return
	160	+
	161	+ # ── 首轮路径：DefaultRequestHandler 未创建 Task 时由 Executor 创建 ──
	162	+ if current_task is None:
	163	+ new_task = Task(
	164	+ id=task_id,
	165	+ context_id=conv_id,
	166	+ status=TaskStatus(state=TASK_STATE_WORKING),
	167	+ )
	168	+ await self._task_store.save(new_task, call_context)
	169	+ logger.debug(f"[Executor] 创建 Task：task={task_id}, conv={conv_id}")
	170	+
	171	+ await self.run_agent(
	172	+ turn_ctx,
	173	+ query=user_query,
	174	+ original_body=original_body,
	175	+ cascade_result=None,
	176	+ step_counter=[0], # cascade 递归共享同一计数器
	177	+ )
	178	+
	179	+ async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
	180	+ pass
	181	+
	182	+ # ── 核心递归编排 ──────────────────────────────────────────────────────────
	183	+
	184	+ async def run_agent(
	185	+ self,
	186	+ turn_ctx: _TurnContext,
	187	+ query: str,
	188	+ original_body: dict,
	189	+ cascade_result: Optional[dict],
	190	+ step_counter: Optional[list[int]] = None,
	191	+ ) -> None:
	192	+ if step_counter is None:
	193	+ step_counter = [0]
	194	+
	195	+ conv_id = turn_ctx.conv_id
	196	+ task_id = turn_ctx.task_id
	197	+ call_context = turn_ctx.call_context
	198	+ event_queue = turn_ctx.event_queue
	199	+
	200	+ turn_start = time.monotonic()
	201	+ is_cascade = cascade_result is not None
	202	+ logger.info(
	203	+ f"[Executor] run_agent 开始: conv={conv_id}, task={task_id}, "
	204	+ f"is_cascade={is_cascade}, step_counter={step_counter[0]}"
	205	+ )
	206	+
	207	+ event_count = 0
	208	+ async for event in agent_stream(
	209	+ query=query,
	210	+ conv_id=conv_id,
	211	+ cascade_result=cascade_result,
	212	+ context={"body": original_body},
	213	+ ):
	214	+ event_count += 1
	215	+ logger.debug(
	216	+ f"[Executor] received agent event #{event_count}: "
	217	+ f"type={type(event).__name__}"
	218	+ )
	219	+ # ── step 边界发射 planning_execution_process ──────────────
	220	+ # 规则：ToolStartEvent 与 DelegateRequest 都计为一个"步骤"
	221	+ if isinstance(event, ToolStartEvent):
	222	+ step_counter[0] += 1
	223	+ desc = event.content or event.plugin or ""
	224	+ planning_content = (
	225	+ f"[执行轨迹] 正在执行步骤{step_counter[0]}: {desc} "
	226	+ f"(tool={event.plugin})"
	227	+ )
	228	+ planning = PlanningExecutionProcessEvent(content=planning_content)
	229	+ planning_a2a = agent_event_to_a2a(planning, task_id, conv_id)
	230	+ if planning_a2a is not None:
	231	+ await event_queue.enqueue_event(planning_a2a)
	232	+ logger.info(
	233	+ f"[Executor] step 边界: 步骤{step_counter[0]} "
	234	+ f"(tool={event.plugin}, desc={desc!r:.60})"
	235	+ )
	236	+
	237	+ if isinstance(event, DelegateRequest):
	238	+ step_counter[0] += 1
	239	+ planning_content = (
	240	+ f"[执行轨迹] 正在执行步骤{step_counter[0]}: "
	241	+ f"{event.task_description} "
	242	+ f"(tool=adapter:versatile_proxy)"
	243	+ )
	244	+ planning = PlanningExecutionProcessEvent(content=planning_content)
	245	+ planning_a2a = agent_event_to_a2a(planning, task_id, conv_id)
	246	+ if planning_a2a is not None:
	247	+ await event_queue.enqueue_event(planning_a2a)
	248	+
	249	+ logger.info(
	250	+ f"[Executor] step 边界: 步骤{step_counter[0]} "
	251	+ f"(tool=adapter:versatile_proxy, intent={event.intent})"
	252	+ )
	253	+ logger.info(
	254	+ f"[Executor] DelegateRequest → {event.intent}: "
	255	+ f"{event.task_description!r:.60}"
	256	+ )
	257	+ va_result, va_task_id = await self._call_versatile_adapter(
	258	+ turn_ctx,
	259	+ delegate=event,
	260	+ )
	261	+ if va_result is not None:
	262	+ await self.run_agent(
	263	+ turn_ctx,
	264	+ query=query,
	265	+ original_body=original_body,
	266	+ cascade_result=va_result,
	267	+ step_counter=step_counter,
	268	+ )
	269	+ else:
	270	+ # VA 未完成：将 va_task_id 写入 Task metadata，状态改为 INPUT_REQUIRED
	271	+ task = await self._task_store.get(task_id, call_context)
	272	+ if task:
	273	+ task.metadata.update({"va_task_id": va_task_id or ""})
	274	+ task.status.CopyFrom(
	275	+ TaskStatus(state=TASK_STATE_INPUT_REQUIRED)
	276	+ )
	277	+ await self._task_store.save(task, call_context)
	278	+ await event_queue.enqueue_event(
	279	+ TaskStatusUpdateEvent(
	280	+ task_id=task_id,
	281	+ context_id=conv_id,
	282	+ status=TaskStatus(state=TASK_STATE_INPUT_REQUIRED),
	283	+ )
	284	+ )
	285	+ logger.info(
	286	+ f"[Executor] VA 挂起：conv={conv_id}, va_task={va_task_id}"
	287	+ )
	288	+ # 记录 VA 挂起 / cascade 路径结束时的累计耗时
	289	+ turn_duration_ms = (time.monotonic() - turn_start) * 1000
	290	+ logger.info(
	291	+ f"[Executor] ⏱️ run_agent 返回: conv={conv_id}, "
	292	+ f"duration={turn_duration_ms:.2f}ms, "
	293	+ f"events_received={event_count}, steps={step_counter[0]}"
	294	+ )
	295	+ return
	296	+
	297	+ a2a_event = agent_event_to_a2a(event, task_id, conv_id)
	298	+ if a2a_event:
	299	+ await event_queue.enqueue_event(a2a_event)
	300	+
	301	+ # agent stream 正常结束（非 DelegateRequest 路径）→ 写 COMPLETED 到 TaskStore
	302	+ task = await self._task_store.get(task_id, call_context)
	303	+ if task and task.status.state != TASK_STATE_COMPLETED:
	304	+ task.status.CopyFrom(TaskStatus(state=TASK_STATE_COMPLETED))
	305	+ await self._task_store.save(task, call_context)
	306	+ logger.debug(f"[Executor] Task 标记 COMPLETED：task={task_id}, conv={conv_id}")
	307	+
	308	+ # 本轮（或本次 cascade）正常结束，打总耗时
	309	+ turn_duration_ms = (time.monotonic() - turn_start) * 1000
	310	+ logger.info(
	311	+ f"[Executor] ⏱️ run_agent 正常结束: conv={conv_id}, "
	312	+ f"duration={turn_duration_ms:.2f}ms, "
	313	+ f"events_received={event_count}, steps_accumulated={step_counter[0]}"
	314	+ )
	315	+
	316	+ # ── VersatileAdapter 调用 ─────────────────────────────────────────────────
	317	+
	318	+ def _build_va_message(self, payload: _VaRequestPayload) -> SendMessageRequest:
	319	+ text_part = Part()
	320	+ text_part.text = payload.query
	321	+
	322	+ data_struct = Struct()
	323	+ data_struct.update(
	324	+ {
	325	+ "headers": payload.headers,
	326	+ "body": payload.body,
	327	+ "params": payload.params or {},
	328	+ }
	329	+ )
	330	+ data_value = Value()
	331	+ data_value.struct_value.CopyFrom(data_struct)
	332	+ data_part = Part()
	333	+ data_part.data.CopyFrom(data_value)
	334	+
	335	+ msg = Message(
	336	+ role=ROLE_USER,
	337	+ message_id=str(uuid.uuid4()),
	338	+ task_id=payload.task_id,
	339	+ context_id=payload.conv_id,
	340	+ )
	341	+ msg.parts.extend([text_part, data_part])
	342	+ return SendMessageRequest(message=msg)
	343	+
	344	+ def _parse_stream_event(self, stream_resp):
	345	+ which = (
	346	+ stream_resp.WhichOneof("payload")
	347	+ if hasattr(stream_resp, "WhichOneof")
	348	+ else None
	349	+ )
	350	+ if which == "artifact_update":
	351	+ return stream_resp.artifact_update
	352	+ if which == "status_update":
	353	+ return stream_resp.status_update
	354	+ return None
	355	+
	356	+ def _extract_node_data(
	357	+ self, event: TaskArtifactUpdateEvent
	358	+ ) -> Optional[dict]:
	359	+ """从 VersatileAdapter 解包后的 artifact 取出节点数据。
	360	+
	361	+ data part 形状：``{"event": "<kind>", "data": <node_data>}``
	362	+ —— 只对 ``event == "message"`` 的帧返回 node_data，其他（如 "end"）返回 None。
	363	+ """
	364	+ for part in event.artifact.parts:
	365	+ if part.WhichOneof("content") == "data":
	366	+ frame = MessageToDict(part.data)
	367	+ if not isinstance(frame, dict):
	368	+ continue
	369	+ if frame.get("event") != "message":
	370	+ continue
	371	+ inner = frame.get("data")
	372	+ if isinstance(inner, dict):
	373	+ return inner
	374	+ return None
	375	+
	376	+ def _extract_end_node(self, event: TaskArtifactUpdateEvent) -> Optional[dict]:
	377	+ node = self._extract_node_data(event)
	378	+ if node is not None and node.get("node_type") == "End":
	379	+ return node
	380	+ return None
	381	+
	382	+ def _is_suppressed_node(self, event: TaskArtifactUpdateEvent) -> bool:
	383	+ """判断该 artifact 是否为配置中需要屏蔽的节点（不推送给用户）。"""
	384	+ target = get_settings().va_workflow_result_node
	385	+ if not target:
	386	+ return False
	387	+ node = self._extract_node_data(event)
	388	+ return node is not None and node.get("node_name") == target
	389	+
	390	+ def _extract_qa_node(self, event: TaskArtifactUpdateEvent) -> Optional[str]:
	391	+ target_node = get_settings().va_workflow_result_node
	392	+ if not target_node:
	393	+ return None
	394	+ node = self._extract_node_data(event)
	395	+ if node is None:
	396	+ return None
	397	+ if node.get("node_type") == "QA" and node.get("node_name") == target_node:
	398	+ return node.get("text", "") or None
	399	+ return None
	400	+
	401	+ async def _call_versatile_adapter(
	402	+ self,
	403	+ turn_ctx: _TurnContext,
	404	+ delegate: DelegateRequest,
	405	+ ) -> tuple[Optional[dict], Optional[str]]:
	406	+ """DPA 委托场景：从 Redis 取首轮缓存，替换 query/intent 后发给 VA。"""
	407	+ conv_id = turn_ctx.conv_id
	408	+ event_queue = turn_ctx.event_queue
	409	+
	410	+ cached = await self._redis.get_json(session_request_key(conv_id)) or {}
	411	+ headers = cached.get("headers", {})
	412	+ body = dict(cached.get("body", {}))
	413	+ params = cached.get("params", {})
	414	+
	415	+ effective_intent, effective_query = _rewrite_recommend_delegate(
	416	+ delegate.intent,
	417	+ delegate.task_description,
	418	+ )
	419	+ if effective_intent != delegate.intent or effective_query != delegate.task_description:
	420	+ logger.info(
	421	+ "[Executor] 推荐入口临时改写：intent={} -> {}, query={!r} -> {!r}",
	422	+ delegate.intent,
	423	+ effective_intent,
	424	+ delegate.task_description,
	425	+ effective_query,
	426	+ )
	427	+
	428	+ input_section = dict(body.get("input") or {})
	429	+ input_section["query"] = effective_query
	430	+ input_section["intent"] = effective_intent
	431	+ body["input"] = input_section
	432	+
	433	+ custom_data = dict(body.get("custom_data") or {})
	434	+ custom_inputs = dict(custom_data.get("inputs") or {})
	435	+ custom_inputs["query"] = effective_query
	436	+ custom_inputs["intent"] = effective_intent
	437	+ custom_data["inputs"] = custom_inputs
	438	+ body["custom_data"] = custom_data
	439	+
	440	+ body["stream"] = True
	441	+
	442	+ # 在 a2a 调用侧记录 Versatile 前后 Tag 日志
	443	+ versatile_call_id = str(uuid.uuid4())
	444	+ versatile_name = get_settings().versatile_adapter_url or "versatile_adapter"
	445	+ call_started_ms = int(time.time() * 1000)
	446	+ status_message = 0
	447	+ error_message: Optional[str] = None
	448	+
	449	+ va_real_task_id: Optional[str] = None
	450	+ continuation_task_id = ""
	451	+
	452	+ request = self._build_va_message(
	453	+ _VaRequestPayload(
	454	+ query=effective_query,
	455	+ headers=headers,
	456	+ body=body,
	457	+ params=params,
	458	+ task_id="",
	459	+ conv_id=conv_id,
	460	+ )
	461	+ )
	462	+
	463	+ has_end_node = False
	464	+ final_result: dict \| None = None
	465	+ qa_result: Optional[str] = None
	466	+ stream_resp_count = 0
	467	+ forwarded_count = 0
	468	+ suppressed_count = 0
	469	+ logger.info(
	470	+ f"[Executor] [VersatileProxy] 开始调用 VA: conv={conv_id}, "
	471	+ f"intent={delegate.intent}, task_desc={delegate.task_description!r:.60}"
	472	+ )
	473	+
	474	+ # 调用前打点：记录请求头/体快照
	475	+ to_logger(
	476	+ message=build_versatile_start_observation(
	477	+ call_id=versatile_call_id,
	478	+ name=versatile_name,
	479	+ request_headers=headers,
	480	+ request_body=body,
	481	+ ),
	482	+ extra=Extra(tag=Tag.TAG_VERSATILE_START, cost=0),
	483	+ )
	484	+
	485	+ try:
	486	+ async for stream_resp in self._va_client.send_message(request):
	487	+ stream_resp_count += 1
	488	+ event = self._parse_stream_event(stream_resp)
	489	+ if event is None:
	490	+ logger.debug(
	491	+ f"[Executor] [VersatileProxy] chunk #{stream_resp_count} "
	492	+ f"解析为 None，跳过"
	493	+ )
	494	+ continue
	495	+
	496	+ if va_real_task_id is None and hasattr(event, "task_id") and event.task_id:
	497	+ va_real_task_id = event.task_id
	498	+ logger.debug(
	499	+ f"[Executor] VA real task_id={va_real_task_id}, conv={conv_id}"
	500	+ )
	501	+
	502	+ if isinstance(event, TaskArtifactUpdateEvent):
	503	+ if self._is_suppressed_node(event):
	504	+ suppressed_count += 1
	505	+ logger.debug(
	506	+ f"[Executor] [VersatileProxy] chunk #{stream_resp_count} "
	507	+ f"命中 va_workflow_result_node，抑制不推送"
	508	+ )
	509	+ else:
	510	+ await event_queue.enqueue_event(event)
	511	+ forwarded_count += 1
	512	+ logger.debug(
	513	+ f"[Executor] [VersatileProxy] chunk #{stream_resp_count} "
	514	+ f"已转发到 event_queue"
	515	+ )
	516	+
	517	+ qa = self._extract_qa_node(event)
	518	+ if qa is not None:
	519	+ qa_result = qa
	520	+ logger.debug(
	521	+ f"[Executor] [VersatileProxy] 提取到 QA 节点 text: "
	522	+ f"{qa!r:.60}"
	523	+ )
	524	+
	525	+ result = self._extract_end_node(event)
	526	+ if result is not None:
	527	+ has_end_node = True
	528	+ final_result = result
	529	+ logger.debug(
	530	+ "[Executor] [VersatileProxy] 检测到 End node，将进入 cascade 路径"
	531	+ )
	532	+
	533	+ except Exception as e:
	534	+ status_message = 1
	535	+ error_message = str(e)
	536	+ logger.exception(f"[Executor] VA send_message 异常：{e}")
	537	+ finally:
	538	+ # 无论成功/异常都补打结束日志，保证调用可观测性完整
	539	+ duration_ms = int(time.time() * 1000) - call_started_ms
	540	+ continuation_task_id = va_real_task_id or str(uuid.uuid4())
	541	+ output_payload = {
	542	+ "stream_resp_count": stream_resp_count,
	543	+ "has_end_node": has_end_node,
	544	+ "va_task_id": continuation_task_id,
	545	+ }
	546	+ if error_message:
	547	+ output_payload["error"] = error_message
	548	+ to_logger(
	549	+ level="ERROR" if status_message else "INFO",
	550	+ message=build_versatile_end_observation(
	551	+ call_id=versatile_call_id,
	552	+ name=versatile_name,
	553	+ output_payload=output_payload,
	554	+ status_message=status_message,
	555	+ duration_ms=duration_ms,
	556	+ ),
	557	+ extra=Extra(tag=Tag.TAG_VERSATILE_END, cost=max(duration_ms, 0)),
	558	+ )
	559	+
	560	+ logger.debug(f"[Executor] VA stream_resp_count={stream_resp_count}, conv={conv_id}")
	561	+
	562	+ if has_end_node:
	563	+ cascade = (
	564	+ {"workflow_result": qa_result} if qa_result is not None else final_result
	565	+ )
	566	+ logger.info(
	567	+ f"[Executor] VA end node: conv={conv_id}, qa_result={qa_result!r:.60}"
	568	+ )
	569	+ return cascade, continuation_task_id
	570	+
	571	+ logger.info(
	572	+ f"[Executor] VA 无 end node: conv={conv_id}, va_task={continuation_task_id}"
	573	+ )
	574	+ return None, continuation_task_id
	575	+
	576	+ async def _continue_versatile_adapter(
	577	+ self,
	578	+ turn_ctx: _TurnContext,
	579	+ va_task_id: str,
	580	+ user_input: str,
	581	+ headers: dict,
	582	+ original_body: dict,
	583	+ ) -> None:
	584	+ """VA 挂起后，下一轮用户输入续轮。va_task_id 为 VA 真实 task_id。"""
	585	+ conv_id = turn_ctx.conv_id
	586	+ task_id = turn_ctx.task_id
	587	+ call_context = turn_ctx.call_context
	588	+ event_queue = turn_ctx.event_queue
	589	+
	590	+ # params 仍从 Redis 首轮缓存取（保留 HEAD 的 params URL query 参数透传）
	591	+ cached = await self._redis.get_json(session_request_key(conv_id)) or {}
	592	+ params = cached.get("params", {})
	593	+ # 对齐 YGQ：续轮直接使用当前请求携带的 body，确保 buyStatus/tranNo 等
	594	+ # 当前轮输入能透传给下游工作流，而不是回退到首轮缓存 body。
	595	+ body = dict(original_body)
	596	+ body["stream"] = True
	597	+
	598	+ # 在 a2a 续轮调用侧记录 Versatile 前后 Tag 日志
	599	+ versatile_call_id = str(uuid.uuid4())
	600	+ versatile_name = get_settings().versatile_adapter_url or "versatile_adapter"
	601	+ call_started_ms = int(time.time() * 1000)
	602	+ status_message = 0
	603	+ error_message: Optional[str] = None
	604	+
	605	+ request = self._build_va_message(
	606	+ _VaRequestPayload(
	607	+ query=user_input,
	608	+ headers=headers,
	609	+ body=body,
	610	+ params=params,
	611	+ task_id=va_task_id,
	612	+ conv_id=conv_id,
	613	+ )
	614	+ )
	615	+
	616	+ has_end_node = False
	617	+ final_result: dict \| None = None
	618	+ qa_result: Optional[str] = None
	619	+ stream_resp_count = 0
	620	+
	621	+ # 续轮调用前打点，记录本次输入上下文
	622	+ to_logger(
	623	+ message=build_versatile_start_observation(
	624	+ call_id=versatile_call_id,
	625	+ name=versatile_name,
	626	+ request_headers=headers,
	627	+ request_body=body,
	628	+ ),
	629	+ extra=Extra(tag=Tag.TAG_VERSATILE_START, cost=0),
	630	+ )
	631	+
	632	+ try:
	633	+ async for stream_resp in self._va_client.send_message(request):
	634	+ stream_resp_count += 1
	635	+ event = self._parse_stream_event(stream_resp)
	636	+ if event is None:
	637	+ continue
	638	+
	639	+ if isinstance(event, TaskArtifactUpdateEvent):
	640	+ if not self._is_suppressed_node(event):
	641	+ await event_queue.enqueue_event(event)
	642	+
	643	+ qa = self._extract_qa_node(event)
	644	+ if qa is not None:
	645	+ qa_result = qa
	646	+
	647	+ result = self._extract_end_node(event)
	648	+ if result is not None:
	649	+ has_end_node = True
	650	+ final_result = result
	651	+
	652	+ except Exception as e:
	653	+ status_message = 1
	654	+ error_message = str(e)
	655	+ logger.exception(f"[Executor] VA continue send_message 异常：{e}")
	656	+ finally:
	657	+ # 续轮结束统一打点，补充状态与耗时
	658	+ duration_ms = int(time.time() * 1000) - call_started_ms
	659	+ output_payload = {
	660	+ "stream_resp_count": stream_resp_count,
	661	+ "has_end_node": has_end_node,
	662	+ "va_task_id": va_task_id,
	663	+ }
	664	+ if error_message:
	665	+ output_payload["error"] = error_message
	666	+ to_logger(
	667	+ level="ERROR" if status_message else "INFO",
	668	+ message=build_versatile_end_observation(
	669	+ call_id=versatile_call_id,
	670	+ name=versatile_name,
	671	+ output_payload=output_payload,
	672	+ status_message=status_message,
	673	+ duration_ms=duration_ms,
	674	+ ),
	675	+ extra=Extra(tag=Tag.TAG_VERSATILE_END, cost=max(duration_ms, 0)),
	676	+ )
	677	+
	678	+ if has_end_node:
	679	+ cascade = (
	680	+ {"workflow_result": qa_result} if qa_result is not None else final_result
	681	+ )
	682	+ logger.info(
	683	+ f"[Executor] VA 续轮 end node: conv={conv_id}, qa_result={qa_result!r:.60}"
	684	+ )
	685	+ # 恢复 Task 到 WORKING 状态后做 cascade 续轮
	686	+ task = await self._task_store.get(task_id, call_context)
	687	+ if task:
	688	+ task.status.CopyFrom(TaskStatus(state=TASK_STATE_WORKING))
	689	+ await self._task_store.save(task, call_context)
	690	+
	691	+ await self.run_agent(
	692	+ turn_ctx,
	693	+ query="",
	694	+ original_body=original_body,
	695	+ cascade_result=cascade,
	696	+ )
	697	+ else:
	698	+ # VA 仍未完成，继续挂起；va_task_id 不变
	699	+ task = await self._task_store.get(task_id, call_context)
	700	+ if task:
	701	+ task.metadata.update({"va_task_id": va_task_id or ""})
	702	+ task.status.CopyFrom(TaskStatus(state=TASK_STATE_INPUT_REQUIRED))
	703	+ await self._task_store.save(task, call_context)
	704	+ await event_queue.enqueue_event(
	705	+ TaskStatusUpdateEvent(
	706	+ task_id=task_id,
	707	+ context_id=conv_id,
	708	+ status=TaskStatus(state=TASK_STATE_INPUT_REQUIRED),
	709	+ )
	710	+ )
	711	+ logger.info(
	712	+ f"[Executor] VA 续轮仍无 end node: conv={conv_id}, va_task={va_task_id}"
	713	+ )

Aapplications/a2a_service/orchestrator/sse_helpers.py+53-0

@@ -0,0 +1,53 @@
	1	+# coding: utf-8
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
	3	+
	4	+"""SSE / EventQueue 辅助函数（无业务依赖，便于独立单测）。"""
	5	+from __future__ import annotations
	6	+
	7	+from typing import Any, Optional
	8	+
	9	+from a2a.server.events.event_queue import QueueShutDown
	10	+from loguru import logger
	11	+
	12	+
	13	+async def next_sse_event(event_queue: Any) -> Optional[Any]:
	14	+ """从 a2a EventQueue 取下一个事件。
	15	+
	16	+ 返回:
	17	+ - 事件对象：队列内还有数据
	18	+ - None：队列已正常关闭且为空（end-of-stream 信号）
	19	+
	20	+ 抛出:
	21	+ 除 ``QueueShutDown`` 之外的任何异常都会上抛，由调用方记 WARNING。
	22	+
	23	+ 注意：a2a-sdk 在 py3.11/3.12 用 ``AsyncQueueShutDown as QueueShutDown`` 别名
	24	+ 导入；类的真实 ``__name__`` 是 ``AsyncQueueShutDown``，所以这里用
	25	+ ``isinstance`` / ``except`` 做类匹配，不能用 ``__name__`` 字符串比较。
	26	+ """
	27	+ try:
	28	+ event = await event_queue.dequeue_event()
	29	+ except QueueShutDown:
	30	+ return None
	31	+ event_queue.task_done()
	32	+ return event
	33	+
	34	+
	35	+def log_outbound_sse(
	36	+ *,
	37	+ conversation_id: str,
	38	+ sequence: int,
	39	+ payload: str,
	40	+ event_kind: str,
	41	+) -> None:
	42	+ """北向 SSE 推送埋点（INFO 级，每帧一行）。
	43	+
	44	+ 用于现场排障：当只能看 a2a_service 日志、看不到客户端抓包时，可以通过这条
	45	+ 日志确认 Runtime 实际向北向推送了哪些事件、字节数与顺序。
	46	+ """
	47	+ logger.info(
	48	+ "[Router] → SSE conv={} #{} kind={} bytes={}",
	49	+ conversation_id,
	50	+ sequence,
	51	+ event_kind,
	52	+ len(payload.encode("utf-8")),
	53	+ )