重构回合-agent-runtime-AtomGit

Rapplications/a2a_service/orchestrator/user_router.py→applications/a2a_service/api/dispatch.py+159-158

	@@ -2,7 +2,7 @@
2	# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved	2	# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3		3
4	"""	4	"""
5	-user_router — 用户入口路由（Versatile 平台定制格式）。	5	+dispatch — 用户入口 HTTP 接线层（Versatile 平台定制格式）。
6		6
7	暴露端点（URL 和参数格式与 Versatile 平台约定保持不变）：	7	暴露端点（URL 和参数格式与 Versatile 平台约定保持不变）：
8	POST /v1/{project_id}/agents/{agent_id}/conversations/{conversation_id}	8	POST /v1/{project_id}/agents/{agent_id}/conversations/{conversation_id}
	@@ -63,19 +63,17 @@ from common.logger import (
63	to_logger,	63	to_logger,
64	get_real_ip, Level, ResultEnum	64	get_real_ip, Level, ResultEnum
65	)	65	)
66	-from common.response_wrapper import (	66	+from channels.normalizer import EventNormalizer
67	- wrap_agent_event,	67	+from channels.mobile_bank_channel import MobileBankChannel
68	- wrap_sub_task_event,	68	+from channels.observability import log_channel_event, record_channel_format_error
69	- wrap_workflow_event,
70	-)
71	from config import get_settings	69	from config import get_settings
72	from orchestrator.executor import Executor	70	from orchestrator.executor import Executor
73	from orchestrator.sse_helpers import log_outbound_sse, next_sse_event	71	from orchestrator.sse_helpers import log_outbound_sse, next_sse_event
		72	+from orchestrator.state import CONV_TASK_KEY
74		73
75	router = APIRouter()	74	router = APIRouter()
76		75
77	_REQUEST_TTL = 1800	76	_REQUEST_TTL = 1800
78	-_CONV_TASK_KEY = "session:{}:a2a_task_id"
79	ERROR_CODE_RATE_LIMIT_EXCEEDED = "100001"	77	ERROR_CODE_RATE_LIMIT_EXCEEDED = "100001"
80	ERROR_MSG_RATE_LIMIT_EXCEEDED = "系统超负载，请在稍后重试"	78	ERROR_MSG_RATE_LIMIT_EXCEEDED = "系统超负载，请在稍后重试"
81	DEFAULT_SESSION_MAX_REQUESTS = 1	79	DEFAULT_SESSION_MAX_REQUESTS = 1
	@@ -315,50 +313,6 @@ async def _check_rate_limit(
315	return True, None, None	313	return True, None, None
316		314
317		315
318	-def _extract_inner_by_kind(inner: dict) -> dict:
319	- """把 sub_task 信封内层（子节点原帧）按 kind 提取为 inner_meta。
320	-
321	- 复用与顶层 agent/workflow 相同的判别，额外处理框架合成的生命周期帧：
322	- - agent 原帧（含 ``type`` 键）→ ``{kind:"agent", type, content, data, plugin}``
323	- - 生命周期 node_start/node_end（``event`` 键）→ ``{kind:"lifecycle", data:<原帧>}``（原样透传）
324	- - workflow VA 原生帧（``event`` + 嵌套 ``data`` dict）→ ``{kind:"workflow", type:<event>, data:<节点数据>}``
325	-
326	- 返回的 inner_meta 交由 ``wrap_sub_task_event`` 渲染为"标准单帧 custom_rsp_data"。
327	- """
328	- if not isinstance(inner, dict):
329	- return {"kind": "lifecycle", "data": {}}
330	-
331	- # agent 原帧：EDPAgent 事件 model_dump，含 type 键（与顶层 agent 分支同构）
332	- if "type" in inner:
333	- nested = inner.get("data")
334	- if isinstance(nested, dict):
335	- payload = nested
336	- else:
337	- payload = {
338	- k: v for k, v in inner.items()
339	- if k not in ("type", "content", "plugin", "data", "display")
340	- }
341	- return {
342	- "kind": "agent",
343	- "type": inner.get("type"),
344	- "content": inner.get("content", "") or "",
345	- "data": payload,
346	- "plugin": inner.get("plugin", ""),
347	- # display（北向 v1.3）贯穿级联：子 Agent think 帧的 display 与顶层一致
348	- "display": inner.get("display"),
349	- }
350	-
351	- event_kind = inner.get("event")
352	- # 框架合成的生命周期帧：原样透传（已是 {event, ...} wire 形态）
353	- if event_kind in ("node_start", "node_end"):
354	- return {"kind": "lifecycle", "data": inner}
355	- # workflow VA 原生帧：{event, data:{...}}（message 等）
356	- if isinstance(inner.get("data"), dict):
357	- return {"kind": "workflow", "type": event_kind, "data": inner["data"]}
358	- # 兜底：按生命周期原样透传，避免误判为 agent 丢失结构
359	- return {"kind": "lifecycle", "data": inner}
360	-
361	-
362	def _extract_event_meta(event) -> Optional[dict]:	316	def _extract_event_meta(event) -> Optional[dict]:
363	"""从 A2A protobuf event 抽出 {kind, type, content, data}。	317	"""从 A2A protobuf event 抽出 {kind, type, content, data}。
364		318
	@@ -367,11 +321,13 @@ def _extract_event_meta(event) -> Optional[dict]:
367	- agent 事件：TaskArtifactUpdateEvent，data part 带 ``type`` 键	321	- agent 事件：TaskArtifactUpdateEvent，data part 带 ``type`` 键
368	→ {"kind": "agent", "type": <event_type>, "content": <text>, "data": <剩余>}	322	→ {"kind": "agent", "type": <event_type>, "content": <text>, "data": <剩余>}
369	- workflow 事件：TaskArtifactUpdateEvent，data part 形状为 ``{"event": <kind>, "data": <dict>}``	323	- workflow 事件：TaskArtifactUpdateEvent，data part 形状为 ``{"event": <kind>, "data": <dict>}``
370	- —— 新 VA（sidecar）经 data_proxy 文本帧下传，a2a 侧 Executor._extract_data_proxy_frames	324	+ —— 由 VersatileAdapter 的 `_unwrap_upstream_frame` 解包后产出；
371	- （json.loads）还原后转发。	325	+ event == "end" 是上游流结束信号，返回 None 由 [DONE] 收尾。
372	- → {"kind": "workflow", "type": "<event>", "content": "", "data": <节点数据>}	326	+ - Completed status events are internal control frames and are not pushed.
373	- - TaskStatusUpdateEvent(COMPLETED) 当作 final_answer_end 对待（agent 事件）	327	+ - Failed status events are mapped to interrupt_start; other status events are skipped.
374	- - 其他 TaskStatusUpdateEvent（如 FAILED）返回 None，不推送给前端	328	+ - TaskStatusUpdateEvent(COMPLETED) 是内部控制状态，不推送给前端；
		329	+ 前端通过 agent 展示帧 final_answer_end 感知回答结束。
		330	+ - FAILED 映射为 interrupt_start，其余 TaskStatusUpdateEvent 不推送。
375	"""	331	"""
376	if isinstance(event, TaskArtifactUpdateEvent):	332	if isinstance(event, TaskArtifactUpdateEvent):
377	text_content = ""	333	text_content = ""
	@@ -383,18 +339,6 @@ def _extract_event_meta(event) -> Optional[dict]:
383	elif kind == "data":	339	elif kind == "data":
384	data_dict = MessageToDict(part.data)	340	data_dict = MessageToDict(part.data)
385		341
386	- # sub_task 信封分支（级联盖章）：在通用 agent/workflow 分支之前判别。
387	- # SubTaskEvent 经 agent_adapter 序列化为 data 带 type="sub_task" 的 artifact。
388	- # 把 sub_task_path/node_kind 提到顶层，内层 data（子节点原帧）按 kind 渲染。
389	- if data_dict.get("type") == "sub_task":
390	- inner_meta = _extract_inner_by_kind(data_dict.get("data") or {})
391	- return {
392	- "kind": "sub_task",
393	- "sub_task_path": data_dict.get("sub_task_path") or [],
394	- "node_kind": data_dict.get("node_kind") or "agent",
395	- "inner": inner_meta,
396	- }
397	-
398	# workflow 事件：VersatileAdapter 解包后的 {event, data} 形状	342	# workflow 事件：VersatileAdapter 解包后的 {event, data} 形状
399	if "event" in data_dict and isinstance(data_dict.get("data"), dict):	343	if "event" in data_dict and isinstance(data_dict.get("data"), dict):
400	event_kind = data_dict["event"]	344	event_kind = data_dict["event"]
	@@ -416,23 +360,17 @@ def _extract_event_meta(event) -> Optional[dict]:
416	# tool_* 事件的 plugin 字段要上浮到 custom_rsp_data.plugin，	360	# tool_* 事件的 plugin 字段要上浮到 custom_rsp_data.plugin，
417	# 避免埋在 data 里再被外层的空串覆盖。	361	# 避免埋在 data 里再被外层的空串覆盖。
418	plugin = data_dict.pop("plugin", "")	362	plugin = data_dict.pop("plugin", "")
419	- # display 字段上浮到 custom_rsp_data.display（北向接口 v1.3），
420	- # 仅当 display 非 None 时才提取，保持向后兼容。
421	- display = data_dict.pop("display", None)
422	# 若事件自带 data 载荷（如 ToolEndEvent.data），直接作为 payload；	363	# 若事件自带 data 载荷（如 ToolEndEvent.data），直接作为 payload；
423	# 否则剩余字段（id/title/status/args/progress 等）即 payload。	364	# 否则剩余字段（id/title/status/args/progress 等）即 payload。
424	nested = data_dict.pop("data", None)	365	nested = data_dict.pop("data", None)
425	payload = nested if isinstance(nested, dict) else data_dict	366	payload = nested if isinstance(nested, dict) else data_dict
426	- meta: dict[str, Any] = {	367	+ return {
427	"kind": "agent",	368	"kind": "agent",
428	"type": event_type,	369	"type": event_type,
429	"content": text_content,	370	"content": text_content,
430	"data": payload,	371	"data": payload,
431	"plugin": plugin,	372	"plugin": plugin,
432	}	373	}
433	- if display is not None:
434	- meta["display"] = display
435	- return meta
436		374
437	# 未知 artifact 形态，回退为 agent thought 兼容	375	# 未知 artifact 形态，回退为 agent thought 兼容
438	logger.debug(	376	logger.debug(
	@@ -448,19 +386,7 @@ def _extract_event_meta(event) -> Optional[dict]:
448		386
449	if isinstance(event, TaskStatusUpdateEvent):	387	if isinstance(event, TaskStatusUpdateEvent):
450	if event.status and event.status.state == TASK_STATE_COMPLETED:	388	if event.status and event.status.state == TASK_STATE_COMPLETED:
451	- content = ""	389	+ return None
452	- if event.status.message:
453	- for part in event.status.message.parts:
454	- if part.WhichOneof("content") == "text":
455	- content = part.text or ""
456	- break
457	- return {
458	- "kind": "agent",
459	- "type": "final_answer_end",
460	- "content": content,
461	- "data": {},
462	- "plugin": "",
463	- }
464	# FAILED 事件映射为 interrupt_start agent 帧，对齐 AgentEngine：	390	# FAILED 事件映射为 interrupt_start agent 帧，对齐 AgentEngine：
465	# 顶层 success=False + error=<msg>，custom_rsp_data.event=interrupt_start。	391	# 顶层 success=False + error=<msg>，custom_rsp_data.event=interrupt_start。
466	# 错误描述放在 status.message.parts[0].text，由 Executor 在 VA 上游报错或	392	# 错误描述放在 status.message.parts[0].text，由 Executor 在 VA 上游报错或
	@@ -487,49 +413,84 @@ def _extract_event_meta(event) -> Optional[dict]:
487	return None	413	return None
488		414
489		415
490	-def _serialize_event(event, *, agent_id: str, conversation_id: str, start_time: float) -> Optional[str]:	416	+def _serialize_event(
		417	+ event,
		418	+ *,
		419	+ agent_id: str,
		420	+ conversation_id: str,
		421	+ start_time: float,
		422	+ channel=None,
		423	+) -> Optional[str]:
491	"""将内部 A2A event 序列化为前端可见的 SSE JSON（已包装）。	424	"""将内部 A2A event 序列化为前端可见的 SSE JSON（已包装）。
492		425
493	返回 None 表示该 event 不需要产出 SSE 帧（调用方应跳过）。	426	返回 None 表示该 event 不需要产出 SSE 帧（调用方应跳过）。
494	"""	427	"""
495	- meta = _extract_event_meta(event)	428	+ normalized = EventNormalizer.normalize(event)
496	- if meta is None:	429	+ if normalized is None:
497	return None	430	return None
498		431
499	elapsed = time.monotonic() - start_time	432	elapsed = time.monotonic() - start_time
500	- if meta["kind"] == "agent":	433	+ formatter = channel or MobileBankChannel()
501	- wrapped = wrap_agent_event(	434	+ try:
502	- event_type=meta["type"],	435	+ wrapped = formatter.format_event(
503	- content=meta["content"],	436	+ normalized,
504	- data=meta["data"],
505	- agent_id=agent_id,
506	- conversation_id=conversation_id,
507	- elapsed=elapsed,
508	- plugin=meta.get("plugin", ""),
509	- success=meta.get("success", True),
510	- error=meta.get("error", ""),
511	- display=meta.get("display"),
512	- )
513	- elif meta["kind"] == "sub_task":
514	- wrapped = wrap_sub_task_event(
515	- sub_task_path=meta["sub_task_path"],
516	- node_kind=meta["node_kind"],
517	- inner_meta=meta["inner"],
518	agent_id=agent_id,	437	agent_id=agent_id,
519	conversation_id=conversation_id,	438	conversation_id=conversation_id,
520	elapsed=elapsed,	439	elapsed=elapsed,
521	)	440	)
522	- else:	441	+ except Exception as exc:
523	- wrapped = wrap_workflow_event(	442	+ record_channel_format_error(
524	- event_kind=meta["type"],	443	+ channel=getattr(formatter, "name", type(formatter).__name__),
525	- data=meta["data"],	444	+ event_type=str(normalized.get("type") or ""),
526	- agent_id=agent_id,	445	+ task_id=getattr(event, "task_id", ""),
527	conversation_id=conversation_id,	446	conversation_id=conversation_id,
528	- elapsed=elapsed,	447	+ error=exc,
529	)	448	)
		449	+ return None
		450	+ if wrapped is None:
		451	+ return None
		452	+ log_channel_event(
		453	+ level=Level.DEBUG,
		454	+ action="CHANNEL_SERIALIZE_EVENT",
		455	+ channel=getattr(formatter, "name", type(formatter).__name__),
		456	+ event_type=str(normalized.get("type") or ""),
		457	+ task_id=getattr(event, "task_id", ""),
		458	+ conversation_id=conversation_id,
		459	+ payload={"agent_id": agent_id, "elapsed": elapsed},
		460	+ )
530	return json.dumps(wrapped, ensure_ascii=False)	461	return json.dumps(wrapped, ensure_ascii=False)
531		462
532		463
		464	+def _extract_answer_from_events(events: list[Any]) -> str:
		465	+ final_answer_chunk = ""
		466	+ final_answer_end = ""
		467	+ completed_answer = ""
		468	+
		469	+ for event in events:
		470	+ normalized = EventNormalizer.normalize(event)
		471	+ if normalized:
		472	+ event_type = normalized.get("type")
		473	+ data = normalized.get("data") if isinstance(normalized.get("data"), dict) else {}
		474	+ content = str(data.get("content") or "")
		475	+ if event_type == "final_answer_chunk" and content:
		476	+ final_answer_chunk = content
		477	+ elif event_type == "final_answer_end" and content:
		478	+ final_answer_end = content
		479	+
		480	+ is_completed_status = (
		481	+ isinstance(event, TaskStatusUpdateEvent)
		482	+ and event.status
		483	+ and event.status.state == TASK_STATE_COMPLETED
		484	+ )
		485	+ if is_completed_status and event.status.message:
		486	+ for part in event.status.message.parts:
		487	+ if part.WhichOneof("content") == "text":
		488	+ completed_answer = part.text
		489	+ break
		490	+
		491	+ return final_answer_chunk or final_answer_end or completed_answer
		492	+
		493	+
533	def _extract_query(body: dict) -> str:	494	def _extract_query(body: dict) -> str:
534	if isinstance(body.get("input"), dict):	495	if isinstance(body.get("input"), dict):
535	q = body["input"].get("query", "")	496	q = body["input"].get("query", "")
	@@ -597,7 +558,7 @@ def _is_flow_control_probe(user_query: str) -> bool:
597		558
598		559
599	async def _ensure_task_mapping(redis, conversation_id: str) -> str:	560	async def _ensure_task_mapping(redis, conversation_id: str) -> str:
600	- conv_task_key = _CONV_TASK_KEY.format(conversation_id)	561	+ conv_task_key = CONV_TASK_KEY.format(conversation_id)
601	task_id = await redis.get(conv_task_key)	562	task_id = await redis.get(conv_task_key)
602	if task_id:	563	if task_id:
603	return task_id	564	return task_id
	@@ -619,39 +580,64 @@ def _build_flow_control_probe_response(conversation_id: str, agent_id: str) -> d
619		580
620		581
621	@router.post(	582	@router.post(
622	- "/v1/{project_id}/agents/{agent_id}/conversations/{conversation_id}/cancel",	583	+ "/v1/{channel_path:path}/cancel",
623	response_model=None,	584	response_model=None,
624	)	585	)
625	async def cancel_task(	586	async def cancel_task(
626	- project_id: str,	587	+ channel_path: str,
627	- agent_id: str,
628	- conversation_id: str,
629	request: Request,	588	request: Request,
630	) -> JSONResponse:	589	) -> JSONResponse:
631	- """整会话取消端点（对齐 TECH §4.2 API 级深度取消）。	590	+ adapter_registry = getattr(request.app.state, "adapter_registry", None)
		591	+ route_spec = None
		592	+ path_params: dict[str, str] = {}
		593	+ base_path = request.url.path.removesuffix("/cancel")
		594	+ if adapter_registry is not None:
		595	+ route_spec, path_params = adapter_registry.match_route(base_path)
		596	+ if route_spec is None:
		597	+ return JSONResponse(
		598	+ status_code=status.HTTP_404_NOT_FOUND,
		599	+ content={
		600	+ "success": False,
		601	+ "error": "channel_route_not_found",
		602	+ "message": f"no channel route matched: {base_path}",
		603	+ },
		604	+ )
		605	+
		606	+ conversation_id = str(path_params.get("conversation_id") or path_params.get("conv_id") or "")
632		607
633	- 设置主 Executor 的取消令牌并经 A2A CancelTask 向各子 Agent 深度传播；
634	- 当前无子 Agent 调度（sub_agent_client=None）时仅设置令牌、行为安全。
635	- """
636	executor: Executor = request.app.state.executor	608	executor: Executor = request.app.state.executor
637	await executor.cancel_task(conversation_id)	609	await executor.cancel_task(conversation_id)
638	- logger.info(f"[Router] cancel 请求：conv={conversation_id}, agent={agent_id}")
639	return JSONResponse({"status": "cancel_requested"})	610	return JSONResponse({"status": "cancel_requested"})
640		611
641		612
642	@router.post(	613	@router.post(
643	- "/v1/{project_id}/agents/{agent_id}/conversations/{conversation_id}",	614	+ "/v1/{channel_path:path}",
644	response_model=None,	615	response_model=None,
645	)	616	)
646	async def dispatch(	617	async def dispatch(
647	- project_id: str,	618	+ channel_path: str,
648	- agent_id: str,
649	- conversation_id: str,
650	request: Request,	619	request: Request,
651	) -> StreamingResponse \| JSONResponse:	620	) -> StreamingResponse \| JSONResponse:
652	traceid = str(uuid.uuid4())	621	traceid = str(uuid.uuid4())
653	settings = get_settings()	622	settings = get_settings()
654	request_started_ms = int(time.time() * 1000)	623	request_started_ms = int(time.time() * 1000)
		624	+ adapter_registry = getattr(request.app.state, "adapter_registry", None)
		625	+ route_spec = None
		626	+ path_params: dict[str, str] = {}
		627	+ if adapter_registry is not None:
		628	+ route_spec, path_params = adapter_registry.match_route(request.url.path)
		629	+ if route_spec is None:
		630	+ return JSONResponse(
		631	+ status_code=status.HTTP_404_NOT_FOUND,
		632	+ content={
		633	+ "success": False,
		634	+ "error": "channel_route_not_found",
		635	+ "message": f"no channel route matched: {request.url.path}",
		636	+ },
		637	+ )
		638	+ agent_id = str(path_params.get("agent_id") or "")
		639	+ conversation_id = str(path_params.get("conversation_id") or path_params.get("conv_id") or "")
		640	+
655	# 在工具类中统一完成请求快照解析与公共字段提取	641	# 在工具类中统一完成请求快照解析与公共字段提取
656	http_request_tag_context = await build_http_request_tag_context(	642	http_request_tag_context = await build_http_request_tag_context(
657	request=request,	643	request=request,
	@@ -748,20 +734,51 @@ async def dispatch(
748	)	734	)
749	return JSONResponse(status_code=400, content=response_content)	735	return JSONResponse(status_code=400, content=response_content)
750		736
751	- user_query = _extract_query(body)	737	+ executor: Executor = request.app.state.executor
752	- stream_mode = bool(body.get("stream", True))	738	+ redis = request.app.state.redis
		739	+ task_store = request.app.state.task_store
		740	+ channel = route_spec.channel
		741	+ event_queue = EventQueue()
		742	+ headers = http_request_tag_context.request_headers
		743	+ params = _extract_query_params(request)
		744	+ try:
		745	+ parsed_request = channel.parse_request(
		746	+ body,
		747	+ path_params=path_params,
		748	+ headers=headers,
		749	+ params=params,
		750	+ )
		751	+ except ValueError as e:
		752	+ response_content = {
		753	+ "success": False,
		754	+ "error": "invalid_request",
		755	+ "message": str(e),
		756	+ }
		757	+ return JSONResponse(status_code=400, content=response_content)
		758	+
		759	+ user_query = parsed_request.query
		760	+ stream_mode = parsed_request.stream
		761	+ agent_id = parsed_request.agent_id
		762	+ conversation_id = parsed_request.conversation_id
		763	+
		764	+ log_channel_event(
		765	+ action="CHANNEL_ROUTE_SELECTED",
		766	+ channel=getattr(channel, "name", type(channel).__name__),
		767	+ conversation_id=conversation_id,
		768	+ payload={
		769	+ "route_key": getattr(route_spec, "route_key", ""),
		770	+ "prefix": getattr(route_spec, "prefix", ""),
		771	+ "path_template": getattr(route_spec, "path_template", ""),
		772	+ "request_path": request.url.path,
		773	+ "agent_id": agent_id,
		774	+ },
		775	+ )
753		776
754	logger.info(	777	logger.info(
755	f"[Router] conv={conversation_id} stream={stream_mode} "	778	f"[Router] conv={conversation_id} stream={stream_mode} "
756	f"query={user_query!r:.80} trace={traceid}"	779	f"query={user_query!r:.80} trace={traceid}"
757	)	780	)
758		781
759	- executor: Executor = request.app.state.executor
760	- redis = request.app.state.redis
761	- task_store = request.app.state.task_store
762	- event_queue = EventQueue()
763	- headers = http_request_tag_context.request_headers
764	-
765	# 入口限流：在进入编排执行前统一拦截	782	# 入口限流：在进入编排执行前统一拦截
766	allowed, error_msg, error_code = await _check_rate_limit(	783	allowed, error_msg, error_code = await _check_rate_limit(
767	redis=redis,	784	redis=redis,
	@@ -839,7 +856,6 @@ async def dispatch(
839	)	856	)
840		857
841	# 缓存首轮请求头、请求体和 URL query params，供 VA 委托调用/续轮恢复使用	858	# 缓存首轮请求头、请求体和 URL query params，供 VA 委托调用/续轮恢复使用
842	- params = _extract_query_params(request)
843	cached = await redis.get_json(session_request_key(conversation_id))	859	cached = await redis.get_json(session_request_key(conversation_id))
844	if cached is None:	860	if cached is None:
845	await redis.set_json(	861	await redis.set_json(
	@@ -904,7 +920,7 @@ async def dispatch(
904		920
905	# ── Task 管理：conv_id → task_id 映射 ─────────────────────────────────	921	# ── Task 管理：conv_id → task_id 映射 ─────────────────────────────────
906	call_context = ServerCallContext()	922	call_context = ServerCallContext()
907	- conv_task_key = _CONV_TASK_KEY.format(conversation_id)	923	+ conv_task_key = CONV_TASK_KEY.format(conversation_id)
908	task_id = await redis.get(conv_task_key)	924	task_id = await redis.get(conv_task_key)
909	current_task = None	925	current_task = None
910		926
	@@ -941,9 +957,7 @@ async def dispatch(
941	f"[Router] 并发首轮，复用 task_id={task_id} for conv={conversation_id}"	957	f"[Router] 并发首轮，复用 task_id={task_id} for conv={conversation_id}"
942	)	958	)
943		959
944	- send_request = _build_request(	960	+ send_request = channel.build_message(parsed_request)
945	- conversation_id, user_query, body, params, headers=headers
946	- )
947	ctx = RequestContext(	961	ctx = RequestContext(
948	call_context=call_context,	962	call_context=call_context,
949	request=send_request,	963	request=send_request,
	@@ -1000,6 +1014,7 @@ async def dispatch(
1000	agent_id=agent_id,	1014	agent_id=agent_id,
1001	conversation_id=conversation_id,	1015	conversation_id=conversation_id,
1002	start_time=turn_start,	1016	start_time=turn_start,
		1017	+ channel=channel,
1003	)	1018	)
1004	# None 表示该事件被屏蔽或是流结束信号，不推送	1019	# None 表示该事件被屏蔽或是流结束信号，不推送
1005	if payload is None:	1020	if payload is None:
	@@ -1091,21 +1106,7 @@ async def dispatch(
1091	break	1106	break
1092	await run_task	1107	await run_task
1093		1108
1094	- def _is_completed_task_event(evt: Any) -> bool:	1109	+ answer = _extract_answer_from_events(collected)
1095	- return (
1096	- isinstance(evt, TaskStatusUpdateEvent)
1097	- and evt.status
1098	- and evt.status.state == TASK_STATE_COMPLETED
1099	- and evt.status.message
1100	- )
1101	-
1102	- answer = ""
1103	- for event in collected:
1104	- if _is_completed_task_event(event):
1105	- for part in event.status.message.parts:
1106	- if part.WhichOneof("content") == "text":
1107	- answer = part.text
1108	- break
1109		1110
1110	response_content = {"success": True, "answer": answer}	1111	response_content = {"success": True, "answer": answer}
1111	duration_ms = int(time.time() * 1000) - request_started_ms	1112	duration_ms = int(time.time() * 1000) - request_started_ms
	@@ -1122,4 +1123,4 @@ async def dispatch(
1122	extra=Extra(tag=Tag.TAG_HTTP_REQUEST_END, cost=max(duration_ms, 0)),	1123	extra=Extra(tag=Tag.TAG_HTTP_REQUEST_END, cost=max(duration_ms, 0)),
1123	log_context=http_request_tag_context.log_context,	1124	log_context=http_request_tag_context.log_context,
1124	)	1125	)
1125	- return JSONResponse(content=response_content)	1126	+ return JSONResponse(content=response_content)

Mapplications/a2a_service/app.py+72-8

	@@ -20,6 +20,7 @@ import socket
20	import sys	20	import sys
21	import time	21	import time
22	from contextlib import asynccontextmanager	22	from contextlib import asynccontextmanager
		23	+from pathlib import Path
23	from typing import Any, Optional	24	from typing import Any, Optional
24		25
25	import httpx	26	import httpx
	@@ -34,13 +35,51 @@ from starlette.applications import Starlette
34		35
35	from agents.EDPAgent import initialize	36	from agents.EDPAgent import initialize
36	from common.redis_client import RedisClient	37	from common.redis_client import RedisClient
37	-from common.redis_task_store import RedisTaskStore	38	+from common.redis_task_store import ReadOnlyTaskStore, RedisTaskStore
38	from config import get_settings	39	from config import get_settings
		40	+from channels.registry import AdapterRegistry
		41	+from api.dispatch import router as dispatch_router
39	from orchestrator.executor import Executor	42	from orchestrator.executor import Executor
40	-from orchestrator.user_router import router as user_router	43	+from orchestrator.route import RouteDispatcher
		44	+from orchestrator.state import TaskStateManager
41	from tools.simulate_router.simulate import router as simulate_router	45	from tools.simulate_router.simulate import router as simulate_router
42		46
43	-os.environ['NO_PROXY'] = 'localhost,127.0.0.1'	47	+
		48	+def _load_env_to_environ() -> None:
		49	+ # pydantic-settings 读 .env 只填充 Settings 对象，不写回 os.environ；
		50	+ # 而 httpx(trust_env=True) 与本模块的 NO_PROXY 合并逻辑都读 os.environ。
		51	+ # 这里用 load_dotenv 把 .env 同步进 os.environ，让代理相关变量对 httpx 生效。
		52	+ # override=False：容器 ENV / 真实环境变量优先级更高，.env 仅作兜底。
		53	+ _env_file = Path(__file__).parent / ".env"
		54	+ if not _env_file.exists():
		55	+ return
		56	+ try:
		57	+ from dotenv import load_dotenv
		58	+ load_dotenv(_env_file, override=False)
		59	+ except ImportError:
		60	+ logger.warning("[NO_PROXY] python-dotenv 未安装，跳过 .env → os.environ 同步")
		61	+
		62	+
		63	+def _init_no_proxy() -> None:
		64	+ # 先把 .env 中的代理变量加载进 os.environ（pydantic-settings 不会做这一步）
		65	+ _load_env_to_environ()
		66	+ # 合并 NO_PROXY：保留本地地址兜底（localhost/127.0.0.1），同时并入 env 已有配置，
		67	+ # 使实际场景可通过 .env 的 NO_PROXY 增加需绕过代理的地址。
		68	+ # 大小写同步设置，避免部分库（requests/urllib3）只读小写 no_proxy 导致行为不一致。
		69	+ local_defaults = {"localhost", "127.0.0.1"}
		70	+ existing = os.environ.get("NO_PROXY") or os.environ.get("no_proxy") or ""
		71	+ merged = local_defaults \| {h.strip() for h in existing.split(",") if h.strip()}
		72	+ value = ",".join(sorted(merged))
		73	+ os.environ["NO_PROXY"] = value
		74	+ os.environ["no_proxy"] = value
		75	+ # 调试日志：确认运行阶段是否正确拿到 env 中的 NO_PROXY 配置
		76	+ logger.info(
		77	+ f"[NO_PROXY] env 原始值={existing!r} 合并后={value!r} "
		78	+ f"HTTP_PROXY={os.environ.get('HTTP_PROXY')!r} HTTPS_PROXY={os.environ.get('HTTPS_PROXY')!r}"
		79	+ )
		80	+
		81	+
		82	+_init_no_proxy()
44		83
45		84
46	LOG_FIELD_SEPARATOR = '\x01'	85	LOG_FIELD_SEPARATOR = '\x01'
	@@ -412,28 +451,53 @@ async def lifespan(fastapi_app: FastAPI):
412	# 也不在启动期拉取子 Agent AgentCard。改为把 factory 注入 Executor，由其按 spec.url	451	# 也不在启动期拉取子 Agent AgentCard。改为把 factory 注入 Executor，由其按 spec.url
413	# 懒 create_from_url + 缓存（每 url 一个 client）。某子 Agent 不可达只降级该实体。	452	# 懒 create_from_url + 缓存（每 url 一个 client）。某子 Agent 不可达只降级该实体。
414	task_store = RedisTaskStore(redis, ttl=settings.redis_session_ttl or _TTL)	453	task_store = RedisTaskStore(redis, ttl=settings.redis_session_ttl or _TTL)
415	- executor = Executor(	454	+ state_manager = TaskStateManager(task_store)
		455	+ dpa_card = _build_dpa_card()
		456	+
		457	+ config_path = os.path.join(
		458	+ os.path.dirname(__file__), "orchestrator", "config", "route_config.yaml"
		459	+ )
		460	+ route_dispatcher = RouteDispatcher(
		461	+ state_manager,
		462	+ config_path=config_path,
		463	+ local_agent_names=[dpa_card.name],
		464	+ )
		465	+ route_dispatcher.register_handlers_from_config(
		466	+ state_manager=state_manager,
416	va_client=va_client,	467	va_client=va_client,
417	redis=redis,	468	redis=redis,
418	- task_store=task_store,
419	client_factory=factory,	469	client_factory=factory,
420	max_concurrent_sub_agents=settings.max_concurrent_sub_agents,	470	max_concurrent_sub_agents=settings.max_concurrent_sub_agents,
421	sub_agent_timeout_seconds=settings.sub_agent_timeout_seconds,	471	sub_agent_timeout_seconds=settings.sub_agent_timeout_seconds,
422	max_parallel_workflows_per_agent=settings.max_parallel_workflows_per_agent,	472	max_parallel_workflows_per_agent=settings.max_parallel_workflows_per_agent,
423	workflow_timeout_seconds=settings.workflow_timeout_seconds,	473	workflow_timeout_seconds=settings.workflow_timeout_seconds,
424	max_call_depth=settings.max_call_depth,	474	max_call_depth=settings.max_call_depth,
		475	+ route_dispatcher=route_dispatcher,
		476	+ local_agent_name=dpa_card.name,
425	)	477	)
426		478
427	- dpa_card = _build_dpa_card()	479	+ executor = Executor(
		480	+ redis=redis,
		481	+ route_dispatcher=route_dispatcher,
		482	+ state_manager=state_manager,
		483	+ )
		484	+
		485	+ # SDK DefaultRequestHandler 的 TaskManager 对每个流式事件都调用 task_store.save()，
		486	+ # 用 ReadOnlyTaskStore 包装使 SDK 的 save 为空操作，避免数千次全量 Task 序列化写入 Redis。
		487	+ # Task 状态持久化由应用层 TaskStateManager 统一管理。
		488	+ sdk_task_store = ReadOnlyTaskStore(task_store)
428	request_handler = DefaultRequestHandler(	489	request_handler = DefaultRequestHandler(
429	agent_executor=executor,	490	agent_executor=executor,
430	- task_store=task_store,	491	+ task_store=sdk_task_store,
431	agent_card=dpa_card,	492	agent_card=dpa_card,
432	)	493	)
433		494
434	fastapi_app.state.redis = redis	495	fastapi_app.state.redis = redis
435	fastapi_app.state.task_store = task_store	496	fastapi_app.state.task_store = task_store
436	fastapi_app.state.executor = executor	497	fastapi_app.state.executor = executor
		498	+ fastapi_app.state.adapter_registry = AdapterRegistry.from_yaml(
		499	+ os.path.join(os.path.dirname(__file__), "channels.yaml")
		500	+ )
437		501
438	a2a_routes = create_agent_card_routes(dpa_card) + create_jsonrpc_routes(	502	a2a_routes = create_agent_card_routes(dpa_card) + create_jsonrpc_routes(
439	request_handler, rpc_url="/"	503	request_handler, rpc_url="/"
	@@ -474,7 +538,7 @@ app = FastAPI(
474	lifespan=lifespan,	538	lifespan=lifespan,
475	)	539	)
476		540
477	-app.include_router(user_router)	541	+app.include_router(dispatch_router)
478	app.include_router(simulate_router)	542	app.include_router(simulate_router)
479		543
480		544

Aapplications/a2a_service/channels.yaml+10-0

@@ -0,0 +1,10 @@
	1	+default_route_key: mobile_bank
	2	+default_channel: mobile_bank
	3	+channels:
	4	+ - name: mobile_bank
	5	+ class: channels.mobile_bank_channel.MobileBankChannel
	6	+routes:
	7	+ - route_key: mobile_bank
	8	+ prefix: /v1
	9	+ path_template: /{project_id}/agents/{agent_id}/conversations/{conversation_id}
	10	+ channel: mobile_bank

Aapplications/a2a_service/channels/__init__.py+11-0

@@ -0,0 +1,11 @@
	1	+from .base import Channel, ParsedRequest
	2	+from .normalizer import EventNormalizer
	3	+from .registry import AdapterRegistry, RouteSpec
	4	+
	5	+__all__ = [
	6	+ "AdapterRegistry",
	7	+ "Channel",
	8	+ "EventNormalizer",
	9	+ "ParsedRequest",
	10	+ "RouteSpec",
	11	+]

Aapplications/a2a_service/channels/base.py+49-0

@@ -0,0 +1,49 @@
	1	+from __future__ import annotations
	2	+
	3	+from abc import ABC, abstractmethod
	4	+from dataclasses import dataclass, field
	5	+from typing import Any
	6	+
	7	+from a2a.types.a2a_pb2 import SendMessageRequest
	8	+
	9	+
	10	+@dataclass(frozen=True)
	11	+class ParsedRequest:
	12	+ conversation_id: str
	13	+ agent_id: str
	14	+ query: str
	15	+ body: dict[str, Any]
	16	+ headers: dict[str, Any] = field(default_factory=dict)
	17	+ params: dict[str, Any] = field(default_factory=dict)
	18	+ stream: bool = True
	19	+ trace_id: str = ""
	20	+
	21	+
	22	+class Channel(ABC):
	23	+ name: str = ""
	24	+
	25	+ @abstractmethod
	26	+ def parse_request(
	27	+ self,
	28	+ body: dict[str, Any],
	29	+ *,
	30	+ path_params: dict[str, Any],
	31	+ headers: dict[str, Any] \| None = None,
	32	+ params: dict[str, Any] \| None = None,
	33	+ ) -> ParsedRequest:
	34	+ """Parse a northbound HTTP request into the channel-neutral request model."""
	35	+
	36	+ @abstractmethod
	37	+ def build_message(self, parsed: ParsedRequest) -> SendMessageRequest:
	38	+ """Build the A2A SendMessageRequest consumed by the orchestrator."""
	39	+
	40	+ @abstractmethod
	41	+ def format_event(
	42	+ self,
	43	+ event: dict[str, Any],
	44	+ *,
	45	+ agent_id: str,
	46	+ conversation_id: str,
	47	+ elapsed: float,
	48	+ ) -> dict[str, Any] \| None:
	49	+ """Format a normalized {type, data} event for the northbound API."""

Aapplications/a2a_service/channels/dict_to_a2a.py+111-0

@@ -0,0 +1,111 @@
	1	+from __future__ import annotations
	2	+
	3	+import uuid
	4	+from typing import Any, Dict, Optional
	5	+
	6	+from a2a.types.a2a_pb2 import Artifact, Part, TaskArtifactUpdateEvent
	7	+from google.protobuf.struct_pb2 import Struct, Value
	8	+
	9	+from common.logger import Level
	10	+from .observability import log_channel_event, record_dict_to_a2a_validation_error
	11	+
	12	+
	13	+# 按 (task_id, event_type) 缓存 artifact_id，使同类流式事件追加到同一 artifact。
	14	+# 避免每个事件都创建新 artifact 导致 Task 体积线性膨胀、save 代价 O(N²) 雪崩。
	15	+# 单个 run_agent 是单协程顺序处理事件流，同一 (task_id, event_type) 不会并发访问。
	16	+_ARTIFACT_ID_CACHE: Dict[str, str] = {}
	17	+
	18	+
	19	+def _resolve_artifact_id(task_id: str, event_type: str) -> tuple[str, bool]:
	20	+ """返回 (artifact_id, append)。
	21	+
	22	+ 首次访问生成新 UUID 并缓存，append=False 让 SDK 创建 artifact；
	23	+ 后续访问复用同一 UUID，append=True 让 SDK 追加 parts 到同一 artifact。
	24	+ """
	25	+ cache_key = f"{task_id}:{event_type}"
	26	+ artifact_id = _ARTIFACT_ID_CACHE.get(cache_key)
	27	+ if artifact_id is None:
	28	+ artifact_id = str(uuid.uuid4())
	29	+ _ARTIFACT_ID_CACHE[cache_key] = artifact_id
	30	+ return artifact_id, False
	31	+ return artifact_id, True
	32	+
	33	+
	34	+def clear_artifact_id_cache(task_id: str) -> None:
	35	+ """清理指定 task_id 相关的 artifact_id 缓存。
	36	+
	37	+ 在 Task 完成/取消/失败时调用，避免缓存随 task 数线性增长。
	38	+ """
	39	+ prefix = f"{task_id}:"
	40	+ keys_to_remove = [k for k in _ARTIFACT_ID_CACHE if k.startswith(prefix)]
	41	+ for k in keys_to_remove:
	42	+ del _ARTIFACT_ID_CACHE[k]
	43	+
	44	+
	45	+def _build_data_part(data: dict[str, Any]) -> Part:
	46	+ struct = Struct()
	47	+ struct.update(data)
	48	+ value = Value()
	49	+ value.struct_value.CopyFrom(struct)
	50	+ part = Part()
	51	+ part.data.CopyFrom(value)
	52	+ return part
	53	+
	54	+
	55	+def dict_to_a2a(
	56	+ event: dict[str, Any],
	57	+ task_id: str,
	58	+ conv_id: str,
	59	+ artifact_id: Optional[str] = None,
	60	+) -> TaskArtifactUpdateEvent:
	61	+ if not isinstance(event, dict):
	62	+ error = TypeError("event must be a dict")
	63	+ record_dict_to_a2a_validation_error(
	64	+ event_type="<unknown>", task_id=task_id, conversation_id=conv_id, error=error
	65	+ )
	66	+ raise error
	67	+ event_type = event.get("type")
	68	+ if not event_type:
	69	+ error = ValueError("event dict must include type")
	70	+ record_dict_to_a2a_validation_error(
	71	+ event_type="<missing>", task_id=task_id, conversation_id=conv_id, error=error
	72	+ )
	73	+ raise error
	74	+ data = event.get("data", {})
	75	+ if data is None:
	76	+ data = {}
	77	+ if not isinstance(data, dict):
	78	+ error = TypeError("event data must be a dict")
	79	+ record_dict_to_a2a_validation_error(
	80	+ event_type=str(event_type), task_id=task_id, conversation_id=conv_id, error=error
	81	+ )
	82	+ raise error
	83	+
	84	+ frame = {"type": event_type, **data}
	85	+ log_channel_event(
	86	+ level=Level.DEBUG,
	87	+ action="DICT_TO_A2A_CONVERT",
	88	+ event_type=str(event_type),
	89	+ task_id=task_id,
	90	+ conversation_id=conv_id,
	91	+ payload={"data_keys": sorted(data.keys())},
	92	+ )
	93	+ # 默认按 (task_id, event_type) 复用 artifact_id，将同类流式事件合并为单个 artifact，
	94	+ # 使 artifacts 数量从 O(事件数) 降为 O(事件类型数)。
	95	+ # 显式传入 artifact_id 时（如测试或需要隔离的场景）直接使用，不参与缓存且 append=False。
	96	+ if artifact_id is None:
	97	+ artifact_id, append = _resolve_artifact_id(task_id, str(event_type))
	98	+ else:
	99	+ append = False
	100	+ artifact = Artifact(
	101	+ artifact_id=artifact_id,
	102	+ parts=[_build_data_part(frame)],
	103	+ )
	104	+ artifact.metadata.update({"type": event_type})
	105	+ return TaskArtifactUpdateEvent(
	106	+ task_id=task_id,
	107	+ context_id=conv_id,
	108	+ artifact=artifact,
	109	+ append=append,
	110	+ last_chunk=False,
	111	+ )

Aapplications/a2a_service/channels/mobile_bank_channel.py+230-0

@@ -0,0 +1,230 @@
	1	+from __future__ import annotations
	2	+
	3	+import uuid
	4	+from typing import Any
	5	+
	6	+from a2a.types.a2a_pb2 import Message, SendMessageRequest
	7	+from google.protobuf.struct_pb2 import Struct, Value
	8	+
	9	+from common.response_wrapper import wrap_agent_event, wrap_workflow_event, wrap_sub_task_event
	10	+from .base import Channel, ParsedRequest
	11	+from .observability import log_channel_event
	12	+
	13	+
	14	+class MobileBankChannel(Channel):
	15	+ name = "mobile_bank"
	16	+
	17	+ def parse_request(
	18	+ self,
	19	+ body: dict[str, Any],
	20	+ *,
	21	+ path_params: dict[str, Any],
	22	+ headers: dict[str, Any] \| None = None,
	23	+ params: dict[str, Any] \| None = None,
	24	+ ) -> ParsedRequest:
	25	+ if not isinstance(body, dict):
	26	+ raise ValueError("request body must be a dict")
	27	+
	28	+ conversation_id = str(
	29	+ path_params.get("conversation_id")
	30	+ or path_params.get("conv_id")
	31	+ or body.get("conversation_id")
	32	+ or ""
	33	+ )
	34	+ agent_id = str(path_params.get("agent_id") or body.get("agent_id") or "")
	35	+ if not conversation_id:
	36	+ raise ValueError("conversation_id is required")
	37	+ if not agent_id:
	38	+ raise ValueError("agent_id is required")
	39	+
	40	+ parsed = ParsedRequest(
	41	+ conversation_id=conversation_id,
	42	+ agent_id=agent_id,
	43	+ query=self._extract_query(body),
	44	+ body=body,
	45	+ headers=headers or {},
	46	+ params=params or {},
	47	+ stream=bool(body.get("stream", True)),
	48	+ trace_id=self._extract_trace_id(body, headers or {}),
	49	+ )
	50	+ log_channel_event(
	51	+ action="CHANNEL_PARSE_REQUEST",
	52	+ channel=self.name,
	53	+ conversation_id=parsed.conversation_id,
	54	+ payload={
	55	+ "agent_id": parsed.agent_id,
	56	+ "stream": parsed.stream,
	57	+ "query_preview": parsed.query[:80],
	58	+ "trace_id": parsed.trace_id,
	59	+ "params_keys": sorted(parsed.params.keys()),
	60	+ "header_keys": sorted(parsed.headers.keys()),
	61	+ },
	62	+ )
	63	+ return parsed
	64	+
	65	+ def build_message(self, parsed: ParsedRequest) -> SendMessageRequest:
	66	+ body_struct = Struct()
	67	+ body_struct.update(
	68	+ {
	69	+ "body": parsed.body,
	70	+ "params": parsed.params,
	71	+ "headers": parsed.headers,
	72	+ }
	73	+ )
	74	+
	75	+ body_value = Value()
	76	+ body_value.struct_value.CopyFrom(body_struct)
	77	+
	78	+ msg = Message()
	79	+ msg.message_id = str(uuid.uuid4())
	80	+ msg.context_id = parsed.conversation_id
	81	+ msg.task_id = ""
	82	+ msg.role = 1
	83	+
	84	+ text_part = msg.parts.add()
	85	+ text_part.text = parsed.query
	86	+
	87	+ data_part = msg.parts.add()
	88	+ data_part.data.CopyFrom(body_value)
	89	+
	90	+ request = SendMessageRequest(message=msg)
	91	+ log_channel_event(
	92	+ action="CHANNEL_BUILD_MESSAGE",
	93	+ channel=self.name,
	94	+ conversation_id=parsed.conversation_id,
	95	+ payload={
	96	+ "agent_id": parsed.agent_id,
	97	+ "message_id": msg.message_id,
	98	+ "parts": len(msg.parts),
	99	+ "has_query": bool(parsed.query),
	100	+ "trace_id": parsed.trace_id,
	101	+ },
	102	+ )
	103	+ return request
	104	+
	105	+ def format_event(
	106	+ self,
	107	+ event: dict[str, Any],
	108	+ *,
	109	+ agent_id: str,
	110	+ conversation_id: str,
	111	+ elapsed: float,
	112	+ ) -> dict[str, Any] \| None:
	113	+ event_type = event.get("type")
	114	+ data = event.get("data") or {}
	115	+ if not event_type or not isinstance(data, dict):
	116	+ log_channel_event(
	117	+ action="CHANNEL_FORMAT_SKIP",
	118	+ channel=self.name,
	119	+ event_type=str(event_type or ""),
	120	+ conversation_id=conversation_id,
	121	+ payload={"reason": "missing_type_or_invalid_data"},
	122	+ )
	123	+ return None
	124	+ if event_type == "versatile_proxy":
	125	+ workflow_event = data.get("event")
	126	+ workflow_data = data.get("data") if isinstance(data.get("data"), dict) else {}
	127	+ if not workflow_event:
	128	+ return None
	129	+ return wrap_workflow_event(
	130	+ event_kind=str(workflow_event),
	131	+ data=workflow_data,
	132	+ agent_id=agent_id,
	133	+ conversation_id=conversation_id,
	134	+ elapsed=elapsed,
	135	+ )
	136	+
	137	+ if event_type == "completed":
	138	+ return None
	139	+
	140	+ if event_type in ("failed", "input_required"):
	141	+ content = str(data.get("content") or data.get("error") or "")
	142	+ return wrap_agent_event(
	143	+ event_type="interrupt_start",
	144	+ content=content,
	145	+ data={},
	146	+ agent_id=agent_id,
	147	+ conversation_id=conversation_id,
	148	+ elapsed=elapsed,
	149	+ success=event_type != "failed",
	150	+ error=str(data.get("error") or "") if event_type == "failed" else "",
	151	+ )
	152	+
	153	+ if event_type == "sub_task":
	154	+ inner = data.get("data") if isinstance(data.get("data"), dict) else {}
	155	+ node_kind = str(data.get("node_kind") or "agent")
	156	+ inner_meta = self._extract_inner_meta(inner, node_kind=node_kind)
	157	+ return wrap_sub_task_event(
	158	+ sub_task_path=[str(p) for p in data.get("sub_task_path") or []],
	159	+ node_kind=node_kind,
	160	+ inner_meta=inner_meta,
	161	+ agent_id=agent_id,
	162	+ conversation_id=conversation_id,
	163	+ elapsed=elapsed,
	164	+ )
	165	+
	166	+ payload = dict(data)
	167	+ content = str(payload.pop("content", "") or "")
	168	+ plugin = str(payload.pop("plugin", "") or "")
	169	+ nested = payload.pop("data", None)
	170	+ if isinstance(nested, dict):
	171	+ payload = nested
	172	+ return wrap_agent_event(
	173	+ event_type=str(event_type),
	174	+ content=content,
	175	+ data=payload,
	176	+ agent_id=agent_id,
	177	+ conversation_id=conversation_id,
	178	+ elapsed=elapsed,
	179	+ plugin=plugin,
	180	+ )
	181	+
	182	+ @staticmethod
	183	+ def _extract_inner_meta(frame: dict[str, Any], *, node_kind: str) -> dict[str, Any]:
	184	+ if not isinstance(frame, dict):
	185	+ return {"kind": "agent", "type": "thought", "content": "", "data": {}}
	186	+ event_kind = str(frame.get("event") or "")
	187	+ if node_kind == "workflow" and event_kind not in ("node_start", "node_end"):
	188	+ return {
	189	+ "kind": "workflow",
	190	+ "type": event_kind or "message",
	191	+ "data": frame.get("data") if isinstance(frame.get("data"), dict) else frame,
	192	+ }
	193	+ if event_kind in ("node_start", "node_end"):
	194	+ return {"kind": "lifecycle", "data": frame}
	195	+ payload = dict(frame)
	196	+ event_type = str(payload.pop("type", "") or event_kind or "thought")
	197	+ content = str(payload.pop("content", "") or "")
	198	+ plugin = str(payload.pop("plugin", "") or "")
	199	+ return {
	200	+ "kind": "agent",
	201	+ "type": event_type,
	202	+ "content": content,
	203	+ "data": payload,
	204	+ "plugin": plugin,
	205	+ "display": frame.get("display"),
	206	+ }
	207	+
	208	+ @staticmethod
	209	+ def _extract_query(body: dict[str, Any]) -> str:
	210	+ if isinstance(body.get("input"), dict):
	211	+ query = body["input"].get("query", "")
	212	+ if query:
	213	+ return str(query)
	214	+ if isinstance(body.get("custom_data"), dict):
	215	+ inputs = body["custom_data"].get("inputs", {})
	216	+ if isinstance(inputs, dict):
	217	+ return str(inputs.get("query", "") or "")
	218	+ return ""
	219	+
	220	+ @staticmethod
	221	+ def _extract_trace_id(body: dict[str, Any], headers: dict[str, Any]) -> str:
	222	+ for key in ("trace_id", "traceid", "x-trace-id", "x-request-id"):
	223	+ value = body.get(key)
	224	+ if value:
	225	+ return str(value)
	226	+ for key in ("trace_id", "traceid", "x-trace-id", "x-request-id"):
	227	+ value = headers.get(key) or headers.get(key.upper())
	228	+ if value:
	229	+ return str(value)
	230	+ return ""

Aapplications/a2a_service/channels/normalizer.py+105-0

@@ -0,0 +1,105 @@
	1	+from __future__ import annotations
	2	+
	3	+import json
	4	+from typing import Any
	5	+
	6	+from a2a.types.a2a_pb2 import (
	7	+ TaskArtifactUpdateEvent,
	8	+ TaskStatusUpdateEvent,
	9	+ TASK_STATE_FAILED,
	10	+ TASK_STATE_INPUT_REQUIRED,
	11	+)
	12	+from google.protobuf.json_format import MessageToDict
	13	+
	14	+
	15	+class EventNormalizer:
	16	+ @staticmethod
	17	+ def normalize(event: Any) -> dict[str, Any] \| None:
	18	+ if isinstance(event, TaskArtifactUpdateEvent):
	19	+ return EventNormalizer._normalize_artifact(event)
	20	+ if isinstance(event, TaskStatusUpdateEvent):
	21	+ return EventNormalizer._normalize_status(event)
	22	+ return None
	23	+
	24	+ @staticmethod
	25	+ def _normalize_artifact(event: TaskArtifactUpdateEvent) -> dict[str, Any] \| None:
	26	+ text_content = ""
	27	+ data_dict: dict[str, Any] = {}
	28	+ for part in event.artifact.parts:
	29	+ kind = part.WhichOneof("content")
	30	+ if kind == "text":
	31	+ text_content = part.text or ""
	32	+ elif kind == "data":
	33	+ data_dict = MessageToDict(part.data)
	34	+
	35	+ metadata_type = ""
	36	+ if event.artifact.HasField("metadata"):
	37	+ meta = MessageToDict(event.artifact.metadata)
	38	+ metadata_type = str(meta.get("type") or "")
	39	+
	40	+ if metadata_type == "versatile_proxy":
	41	+ return {"type": "versatile_proxy", "data": data_dict}
	42	+
	43	+ if "event" in data_dict and isinstance(data_dict.get("data"), dict):
	44	+ return {
	45	+ "type": "versatile_proxy",
	46	+ "data": {
	47	+ "event": data_dict.get("event"),
	48	+ "data": data_dict.get("data") or {},
	49	+ },
	50	+ }
	51	+
	52	+ event_type = data_dict.get("type") or metadata_type
	53	+ if event_type:
	54	+ frame = dict(data_dict)
	55	+ frame.pop("type", None)
	56	+ if text_content and "content" not in frame:
	57	+ frame["content"] = text_content
	58	+ return {"type": str(event_type), "data": frame}
	59	+
	60	+ if text_content or data_dict:
	61	+ return {
	62	+ "type": "thought",
	63	+ "data": {"content": text_content, **data_dict},
	64	+ }
	65	+ return None
	66	+
	67	+ @staticmethod
	68	+ def _normalize_status(event: TaskStatusUpdateEvent) -> dict[str, Any] \| None:
	69	+ if not event.status:
	70	+ return None
	71	+
	72	+ content = EventNormalizer.extract_status_content(event)
	73	+ data: dict[str, Any] = {}
	74	+ if event.status.message:
	75	+ for part in event.status.message.parts:
	76	+ if part.WhichOneof("content") == "data":
	77	+ maybe_data = MessageToDict(part.data)
	78	+ if isinstance(maybe_data, dict):
	79	+ data = maybe_data
	80	+ break
	81	+ if event.HasField("metadata"):
	82	+ meta = MessageToDict(event.metadata)
	83	+ cascade_result = meta.get("cascade_result")
	84	+ if isinstance(cascade_result, dict):
	85	+ data = cascade_result
	86	+
	87	+ state = event.status.state
	88	+ if state == TASK_STATE_FAILED:
	89	+ return {"type": "failed", "data": {"content": content, "error": content}}
	90	+ if state == TASK_STATE_INPUT_REQUIRED:
	91	+ return {"type": "input_required", "data": {"content": content}}
	92	+ return None
	93	+
	94	+ @staticmethod
	95	+ def extract_status_content(event: TaskStatusUpdateEvent) -> str:
	96	+ if not event.status or not event.status.message:
	97	+ return ""
	98	+ for part in event.status.message.parts:
	99	+ kind = part.WhichOneof("content")
	100	+ if kind == "text":
	101	+ return part.text or ""
	102	+ if kind == "data":
	103	+ data = MessageToDict(part.data)
	104	+ return json.dumps(data, ensure_ascii=False, sort_keys=True) if isinstance(data, dict) else ""
	105	+ return ""

Aapplications/a2a_service/channels/observability.py+127-0

@@ -0,0 +1,127 @@
	1	+from __future__ import annotations
	2	+
	3	+from collections import Counter
	4	+from threading import Lock
	5	+from typing import Any
	6	+
	7	+from common.logger import Extra, Level, Tag, mask_sensitive_fields, to_logger
	8	+
	9	+
	10	+_COUNTERS: Counter[tuple[str, tuple[tuple[str, str], ...]]] = Counter()
	11	+_LOCK = Lock()
	12	+
	13	+
	14	+def increment_counter(name: str, **labels: Any) -> None:
	15	+ normalized_labels = tuple(
	16	+ sorted((str(key), str(value)) for key, value in labels.items())
	17	+ )
	18	+ with _LOCK:
	19	+ _COUNTERS[(name, normalized_labels)] += 1
	20	+
	21	+
	22	+def get_counter_value(name: str, **labels: Any) -> int:
	23	+ normalized_labels = tuple(
	24	+ sorted((str(key), str(value)) for key, value in labels.items())
	25	+ )
	26	+ with _LOCK:
	27	+ return _COUNTERS[(name, normalized_labels)]
	28	+
	29	+
	30	+def reset_counters() -> None:
	31	+ with _LOCK:
	32	+ _COUNTERS.clear()
	33	+
	34	+
	35	+def log_channel_event(
	36	+ *,
	37	+ level: str = Level.INFO,
	38	+ action: str,
	39	+ channel: str = "",
	40	+ event_type: str = "",
	41	+ task_id: str = "",
	42	+ conversation_id: str = "",
	43	+ source_agent: str = "",
	44	+ payload: dict[str, Any] \| None = None,
	45	+) -> None:
	46	+ message = {
	47	+ "action": action,
	48	+ "channel": channel,
	49	+ "event_type": event_type,
	50	+ "task_id": task_id,
	51	+ "conversation_id": conversation_id,
	52	+ "source_agent": source_agent,
	53	+ "payload": mask_sensitive_fields(payload or {}),
	54	+ }
	55	+ to_logger(
	56	+ level=level,
	57	+ message=message,
	58	+ extra=Extra(tag=Tag.TAG_CUSTOM, cost=0),
	59	+ )
	60	+
	61	+
	62	+def record_channel_format_error(
	63	+ *,
	64	+ channel: str,
	65	+ event_type: str,
	66	+ task_id: str = "",
	67	+ conversation_id: str = "",
	68	+ source_agent: str = "",
	69	+ error: Exception \| str,
	70	+) -> None:
	71	+ increment_counter(
	72	+ "a2a_channel_format_errors_total",
	73	+ channel=channel,
	74	+ event_type=event_type,
	75	+ )
	76	+ log_channel_event(
	77	+ level=Level.WARNING,
	78	+ action="A2A_WARNING:CHANNEL_FORMAT_ERROR",
	79	+ channel=channel,
	80	+ event_type=event_type,
	81	+ task_id=task_id,
	82	+ conversation_id=conversation_id,
	83	+ source_agent=source_agent,
	84	+ payload={"error": str(error)},
	85	+ )
	86	+
	87	+
	88	+def record_dict_to_a2a_validation_error(
	89	+ *,
	90	+ event_type: str,
	91	+ task_id: str = "",
	92	+ conversation_id: str = "",
	93	+ source_agent: str = "",
	94	+ error: Exception \| str,
	95	+) -> None:
	96	+ increment_counter(
	97	+ "a2a_dict_to_a2a_validation_errors_total",
	98	+ event_type=event_type,
	99	+ )
	100	+ log_channel_event(
	101	+ level=Level.WARNING,
	102	+ action="A2A_WARNING:DICT_TO_A2A_VALIDATION_ERROR",
	103	+ event_type=event_type,
	104	+ task_id=task_id,
	105	+ conversation_id=conversation_id,
	106	+ source_agent=source_agent,
	107	+ payload={"error": str(error)},
	108	+ )
	109	+
	110	+
	111	+def record_dict_to_a2a_downgrade(
	112	+ *,
	113	+ event_type: str,
	114	+ task_id: str = "",
	115	+ conversation_id: str = "",
	116	+ source_agent: str = "",
	117	+ reason: str,
	118	+) -> None:
	119	+ log_channel_event(
	120	+ level=Level.WARNING,
	121	+ action="A2A_WARNING:DICT_TO_A2A_DOWNGRADE",
	122	+ event_type=event_type,
	123	+ task_id=task_id,
	124	+ conversation_id=conversation_id,
	125	+ source_agent=source_agent,
	126	+ payload={"reason": reason},
	127	+ )

Aapplications/a2a_service/channels/registry.py+185-0

@@ -0,0 +1,185 @@
	1	+from __future__ import annotations
	2	+
	3	+import importlib
	4	+from dataclasses import dataclass
	5	+from pathlib import Path
	6	+from typing import Any
	7	+
	8	+import yaml
	9	+
	10	+from .base import Channel
	11	+
	12	+
	13	+@dataclass(frozen=True)
	14	+class RouteSpec:
	15	+ route_key: str
	16	+ prefix: str
	17	+ path_template: str
	18	+ channel: Channel
	19	+
	20	+
	21	+class AdapterRegistry:
	22	+ def __init__(self) -> None:
	23	+ self._channels: dict[str, Channel] = {}
	24	+ self._specs: dict[str, RouteSpec] = {}
	25	+ self._default_route_key = "mobile_bank"
	26	+
	27	+ def register_channel(self, name: str, channel: Channel) -> None:
	28	+ if not name:
	29	+ raise ValueError("channel name is required")
	30	+ self._channels[name] = channel
	31	+
	32	+ def register(self, route_key: str, spec: RouteSpec) -> None:
	33	+ if not route_key:
	34	+ raise ValueError("route_key is required")
	35	+ self._specs[route_key] = spec
	36	+ channel_name = getattr(spec.channel, "name", route_key) or route_key
	37	+ self._channels.setdefault(channel_name, spec.channel)
	38	+
	39	+ def get(self, route_key: str \| None = None) -> RouteSpec \| Channel:
	40	+ """Return RouteSpec by route_key; no args keeps legacy default Channel behavior."""
	41	+ if route_key is None:
	42	+ return self.get_channel()
	43	+ return self._specs.get(route_key)
	44	+
	45	+ def get_channel(self, route_key: str \| None = None) -> Channel:
	46	+ key = route_key or self._default_route_key
	47	+ spec = self._specs.get(key)
	48	+ if spec is not None:
	49	+ return spec.channel
	50	+ if key in self._channels:
	51	+ return self._channels[key]
	52	+ raise KeyError(f"route/channel not registered: {key}")
	53	+
	54	+ def all_specs(self) -> dict[str, RouteSpec]:
	55	+ return dict(self._specs)
	56	+
	57	+ def match_path(self, path: str) -> RouteSpec \| None:
	58	+ normalized_path = path or ""
	59	+ for spec in self._specs.values():
	60	+ if _extract_path_params(normalized_path, spec) is not None:
	61	+ return spec
	62	+ return self._specs.get(self._default_route_key)
	63	+
	64	+ def match_route(self, path: str) -> tuple[RouteSpec \| None, dict[str, str]]:
	65	+ """Strictly match a request path and extract template variables.
	66	+
	67	+ Unlike match_path(), this does not fall back to the default route. HTTP
	68	+ entrypoints should use this method so unknown paths fail fast.
	69	+ """
	70	+ normalized_path = path or ""
	71	+ for spec in self._specs.values():
	72	+ params = _extract_path_params(normalized_path, spec)
	73	+ if params is not None:
	74	+ return spec, params
	75	+ return None, {}
	76	+
	77	+ @classmethod
	78	+ def from_config(cls, config: dict[str, Any] \| None) -> "AdapterRegistry":
	79	+ registry = cls()
	80	+ data = config or {}
	81	+
	82	+ default_route_key = (
	83	+ data.get("default_route_key")
	84	+ or data.get("default_channel")
	85	+ or "mobile_bank"
	86	+ )
	87	+ registry._default_route_key = str(default_route_key)
	88	+
	89	+ channel_instances: dict[str, Channel] = {}
	90	+ channels = data.get("channels") or []
	91	+ if not channels:
	92	+ channels = [
	93	+ {
	94	+ "name": "mobile_bank",
	95	+ "class": "channels.mobile_bank_channel.MobileBankChannel",
	96	+ }
	97	+ ]
	98	+
	99	+ for item in channels:
	100	+ name = str(item["name"])
	101	+ channel = _load_channel(str(item["class"]))
	102	+ registry.register_channel(name, channel)
	103	+ channel_instances[name] = channel
	104	+
	105	+ routes = data.get("routes")
	106	+ if routes is None:
	107	+ default_channel = str(data.get("default_channel") or default_route_key)
	108	+ routes = [
	109	+ {
	110	+ "route_key": default_route_key,
	111	+ "prefix": "/v1",
	112	+ "path_template": "/{project_id}/agents/{agent_id}/conversations/{conversation_id}",
	113	+ "channel": default_channel,
	114	+ }
	115	+ ]
	116	+
	117	+ for item in routes:
	118	+ route_key = str(item["route_key"])
	119	+ channel_name = str(item.get("channel") or route_key)
	120	+ channel = channel_instances.get(channel_name) or registry._channels.get(channel_name)
	121	+ if channel is None:
	122	+ raise KeyError(f"channel not registered for route {route_key}: {channel_name}")
	123	+ registry.register(
	124	+ route_key,
	125	+ RouteSpec(
	126	+ route_key=route_key,
	127	+ prefix=str(item.get("prefix") or ""),
	128	+ path_template=str(item.get("path_template") or item.get("path") or ""),
	129	+ channel=channel,
	130	+ ),
	131	+ )
	132	+
	133	+ return registry
	134	+
	135	+ @classmethod
	136	+ def from_yaml(cls, path: str \| Path) -> "AdapterRegistry":
	137	+ config_path = Path(path)
	138	+ data: dict[str, Any] = {}
	139	+ if config_path.exists():
	140	+ data = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
	141	+ return cls.from_config(data)
	142	+
	143	+
	144	+def _load_channel(class_path: str) -> Channel:
	145	+ module_name, class_name = class_path.rsplit(".", 1)
	146	+ module = importlib.import_module(module_name)
	147	+ channel_cls = getattr(module, class_name)
	148	+ return channel_cls()
	149	+
	150	+
	151	+def _match_template(path: str, template: str) -> bool:
	152	+ return _extract_template_params(path, template) is not None
	153	+
	154	+
	155	+def _extract_path_params(path: str, spec: RouteSpec) -> dict[str, str] \| None:
	156	+ if spec.prefix and not path.startswith(spec.prefix):
	157	+ return None
	158	+
	159	+ candidates = [path]
	160	+ if spec.prefix:
	161	+ stripped = path[len(spec.prefix):] or "/"
	162	+ candidates.insert(0, stripped)
	163	+
	164	+ for candidate in candidates:
	165	+ params = _extract_template_params(candidate, spec.path_template)
	166	+ if params is not None:
	167	+ return params
	168	+ return None
	169	+
	170	+
	171	+def _extract_template_params(path: str, template: str) -> dict[str, str] \| None:
	172	+ if not template:
	173	+ return {}
	174	+ path_parts = [part for part in path.strip("/").split("/") if part]
	175	+ template_parts = [part for part in template.strip("/").split("/") if part]
	176	+ if len(path_parts) != len(template_parts):
	177	+ return None
	178	+ params: dict[str, str] = {}
	179	+ for actual, expected in zip(path_parts, template_parts):
	180	+ if expected.startswith("{") and expected.endswith("}"):
	181	+ params[expected[1:-1]] = actual
	182	+ continue
	183	+ if actual != expected:
	184	+ return None
	185	+ return params

Mapplications/a2a_service/common/__init__.py+8-25

	@@ -1,29 +1,12 @@
1	# coding: utf-8	1	# coding: utf-8
2	# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved	2	# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3		3
4	-from .events import (	4	+"""Shared runtime helpers package.
5	- AgentEvent,
6	- ThoughtEvent,
7	- AnswerEvent,
8	- DelegateRequest,
9	- # 并行子 Agent / 多工作流
10	- SubAgentSpec,
11	- SubAgentDispatchRequest,
12	- SubAgentResult,
13	- WorkflowSpec,
14	- MultiDelegateRequest,
15	- SubTaskEvent,
16	-)
17		5
18	-__all__ = [	6	+Event models live with the concrete agent implementation under
19	- "AgentEvent",	7	+``agents.EDPAgent.events``. Keep package initialization lightweight so imports
20	- "ThoughtEvent",	8	+such as ``common.logger`` and ``common.redis_client`` do not pull legacy event
21	- "AnswerEvent",	9	+modules into the runtime path.
22	- "DelegateRequest",	10	+"""
23	- "SubAgentSpec",	11	+
24	- "SubAgentDispatchRequest",	12	+__all__: list[str] = []
25	- "SubAgentResult",
26	- "WorkflowSpec",
27	- "MultiDelegateRequest",
28	- "SubTaskEvent",
29	-]

Dapplications/a2a_service/common/events.py+0-445

	@@ -1,445 +0,0 @@
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 dataclasses import dataclass
25	-from typing import Any, Literal, Optional, Union
26	-
27	-from pydantic import BaseModel, Field
28	-
29	-
30	-# ════════════════════════════════════════════════════════════════════
31	-# 会话事件
32	-# ════════════════════════════════════════════════════════════════════
33	-
34	-
35	-class ConversationStartEvent(BaseModel):
36	- """对话开启。每次北向请求开始时发出一次。"""
37	- type: Literal["conversation_start"] = "conversation_start"
38	- content: str = ""
39	-
40	-
41	-class ConversationEndEvent(BaseModel):
42	- """对话结束。整个处理流程结束时发出一次。"""
43	- type: Literal["conversation_end"] = "conversation_end"
44	- content: str = ""
45	-
46	-
47	-# ════════════════════════════════════════════════════════════════════
48	-# 思考事件（从 llm_reasoning 流中分段）
49	-# ════════════════════════════════════════════════════════════════════
50	-
51	-
52	-class ThinkStartEvent(BaseModel):
53	- """LLM 思考开始（每轮 ReAct 的第一个 reasoning chunk）。"""
54	- type: Literal["think_start"] = "think_start"
55	- content: str = ""
56	- display: bool \| None = Field(
57	- default=None,
58	- description="true=前端开始拼接, false=前端停止拼接, None=不指定(向后兼容)"
59	- )
60	-
61	-
62	-class ThinkChunkEvent(BaseModel):
63	- """LLM 思考流式片段。"""
64	- type: Literal["think_chunk"] = "think_chunk"
65	- content: str
66	- display: bool \| None = Field(
67	- default=None,
68	- description="true=前端继续拼接, false=前端停止拼接, None=不指定(向后兼容)"
69	- )
70	-
71	-
72	-class ThinkEndEvent(BaseModel):
73	- """LLM 思考结束（reasoning 流结束或切换为 tool_call / answer）。"""
74	- type: Literal["think_end"] = "think_end"
75	- content: str = ""
76	- display: bool \| None = Field(
77	- default=None,
78	- description="true=前端停止拼接并展示, false=前端丢弃拼接内容, None=不指定(向后兼容)"
79	- )
80	-
81	-
82	-# ════════════════════════════════════════════════════════════════════
83	-# 规划事件（Todolist，由 LLM 在 thought 中按约定 JSON 输出）
84	-# ════════════════════════════════════════════════════════════════════
85	-
86	-
87	-class TodoListStartEvent(BaseModel):
88	- """Todolist 生成开始。"""
89	- type: Literal["todolist_start"] = "todolist_start"
90	- content: str = ""
91	-
92	-
93	-class TodoListItemEvent(BaseModel):
94	- """Todolist 中的单个任务条目。"""
95	- type: Literal["todolist_item"] = "todolist_item"
96	- id: int \| str
97	- title: str
98	- status: Literal["pending", "in_progress", "done", "failed", "cancelled"] = "pending"
99	- content: str = ""
100	-
101	-
102	-class TodoListEndEvent(BaseModel):
103	- """Todolist 生成完成。"""
104	- type: Literal["todolist_end"] = "todolist_end"
105	- count: int = 0
106	- content: str = ""
107	-
108	-
109	-# ════════════════════════════════════════════════════════════════════
110	-# 任务事件（单个 todo 状态变更，由 LLM 在 thought 中按约定 JSON 输出）
111	-# ════════════════════════════════════════════════════════════════════
112	-
113	-
114	-class TodoStartEvent(BaseModel):
115	- """单个 Todo 开始执行。"""
116	- type: Literal["todo_start"] = "todo_start"
117	- id: int \| str
118	- title: str = ""
119	- content: str = ""
120	-
121	-
122	-class TodoStatusEvent(BaseModel):
123	- """单个 Todo 状态变更（执行中）。"""
124	- type: Literal["todo_status"] = "todo_status"
125	- id: int \| str
126	- status: Literal["pending", "in_progress", "done", "failed"]
127	- content: str = ""
128	-
129	-
130	-class TodoEndEvent(BaseModel):
131	- """单个 Todo 执行结束。"""
132	- type: Literal["todo_end"] = "todo_end"
133	- id: int \| str
134	- status: Literal["done", "failed"] = "done"
135	- content: str = ""
136	-
137	-
138	-# ════════════════════════════════════════════════════════════════════
139	-# 工具事件（从 Runner 的 tool_start / tool_end 映射）
140	-# ════════════════════════════════════════════════════════════════════
141	-
142	-
143	-class ToolStartEvent(BaseModel):
144	- """工具调用开始。"""
145	- type: Literal["tool_start"] = "tool_start"
146	- content: str = ""
147	- plugin: str = Field(default="", description="工具名")
148	- args: dict[str, Any] = Field(default_factory=dict, description="工具入参")
149	-
150	-
151	-class ToolStatusEvent(BaseModel):
152	- """工具调用进行中（可选，用于长时任务的进度）。"""
153	- type: Literal["tool_status"] = "tool_status"
154	- plugin: str = ""
155	- content: str = ""
156	- progress: Optional[float] = None # 0.0 - 1.0
157	-
158	-
159	-class ToolEndEvent(BaseModel):
160	- """工具调用结束。"""
161	- type: Literal["tool_end"] = "tool_end"
162	- content: str = ""
163	- plugin: str = Field(default="", description="工具名")
164	- data: dict[str, Any] = Field(default_factory=dict, description="工具返回数据")
165	-
166	-
167	-# ════════════════════════════════════════════════════════════════════
168	-# 执行轨迹事件（由 Executor 在 step 边界发射）
169	-# ════════════════════════════════════════════════════════════════════
170	-
171	-
172	-class PlanningExecutionProcessEvent(BaseModel):
173	- """执行轨迹（step 边界标记）。
174	-
175	- 形如 "[执行轨迹] 正在执行步骤N: <desc> (tool=<tool_name>)" 的字符串在 content
176	- 里提供给前端做"当前卡在哪一步"的展示。
177	-
178	- 此事件在北向包装里独有 error_code: "" 字段（对齐抓包）。
179	- """
180	- type: Literal["planning_execution_process"] = "planning_execution_process"
181	- content: str = ""
182	-
183	-
184	-# ════════════════════════════════════════════════════════════════════
185	-# 中断事件（HITL）
186	-# ════════════════════════════════════════════════════════════════════
187	-
188	-
189	-class InterruptStartEvent(BaseModel):
190	- """等待用户输入。"""
191	- type: Literal["interrupt_start"] = "interrupt_start"
192	- interrupt_id: str
193	- content: str = Field(default="", description="提示话术")
194	- context: dict[str, Any] = Field(default_factory=dict)
195	-
196	-
197	-class InterruptEndEvent(BaseModel):
198	- """中断已恢复（用户输入被接受）。"""
199	- type: Literal["interrupt_end"] = "interrupt_end"
200	- interrupt_id: str = ""
201	- content: str = Field(default="", description="回显确认信息")
202	-
203	-
204	-# ════════════════════════════════════════════════════════════════════
205	-# 总结事件（从 llm_output / answer 流分段）
206	-# ════════════════════════════════════════════════════════════════════
207	-
208	-
209	-class FinalAnswerStartEvent(BaseModel):
210	- """最终回答开始。"""
211	- type: Literal["final_answer_start"] = "final_answer_start"
212	- content: str = ""
213	-
214	-
215	-class SummaryEvent(BaseModel):
216	- """流式总结片段（按 LLM token 切片逐段发出）。
217	-
218	- 规范定义（feat-north-api-sse.md §4.5.9 / north-api-response-format.md §4.9）：
219	- - `summary`：流式通道，承担"边想边说"的 UI 展示
220	- - `final_answer_chunk`：一次性全量通道，承担"最终权威文本"
221	- 两者并存，互补不冲突。前端流式展示用 SummaryEvent，存档/回显用
222	- FinalAnswerChunkEvent。
223	- """
224	- type: Literal["summary"] = "summary"
225	- content: str
226	-
227	-
228	-class FinalAnswerChunkEvent(BaseModel):
229	- """最终回答的一次性全量文本帧。
230	-
231	- 注意：抓包与规范一致 —— 这一帧并非流式片段，而是在所有 `summary` 流式
232	- 片段结束后，以单帧形式发送完整版（供校对/回显/存档使用）。
233	-
234	- 流式片段请使用 SummaryEvent。
235	- """
236	- type: Literal["final_answer_chunk"] = "final_answer_chunk"
237	- content: str
238	-
239	-
240	-class FinalAnswerEndEvent(BaseModel):
241	- """最终回答结束。"""
242	- type: Literal["final_answer_end"] = "final_answer_end"
243	- content: str = ""
244	-
245	-
246	-# ════════════════════════════════════════════════════════════════════
247	-# Executor 专用（非北向事件，不出流）
248	-# ════════════════════════════════════════════════════════════════════
249	-
250	-
251	-class DelegateRequest(BaseModel):
252	- """
253	- Agent 需要外部 Agent 处理子任务时 yield 的委托对象。
254	-
255	- 协议约定：Orchestrator 收到此对象后：
256	- 1. 调用 target_agent 完成子任务
257	- 2. 获得 workflow_result（dict）
258	- 3. 以 cascade_result=workflow_result 再次调用 agent_stream()
259	- """
260	- type: Literal["delegate"] = "delegate"
261	- intent: str = Field(description="意图标识")
262	- target_agent: str \| None = None
263	- task_description: str = Field(description="自然语言任务描述")
264	-
265	-
266	-# ════════════════════════════════════════════════════════════════════
267	-# 并行子 Agent / 多工作流（Executor 专用 + 级联盖章信封）
268	-# 对齐 TECH_Design.md §2.1：
269	-# - SubAgentSpec / SubAgentDispatchRequest 主 Agent 侧，派发并行子 Agent
270	-# - WorkflowSpec / MultiDelegateRequest 子 Agent 侧，派发并行工作流
271	-# - SubAgentResult Executor 内部聚合，不出 SSE
272	-# - SubTaskEvent 级联盖章信封，进入 SSE 流、前端消费
273	-# ════════════════════════════════════════════════════════════════════
274	-
275	-
276	-@dataclass
277	-class SubAgentSpec:
278	- """单个并行子 Agent 的派发规格（主 Agent 侧）。"""
279	- entity_id: str # 业务实体唯一标识，同时作为前端显示 key / sub_task_path 段
280	- entity_name: str # LLM 识别的中文公司名称，透传至前端展示
281	- query: str # 针对该实体的子任务描述
282	- url: str = "" # 该子 Agent 的 A2A 端点；由 Agent 自管、派发时下传（P-006）。
283	- # 框架据此按 url 懒构造/缓存 client；空 + 无注入 client → 该实体降级 failed。
284	-
285	-
286	-class SubAgentDispatchRequest(BaseModel):
287	- """主 Agent yield 的并行子 Agent 派发请求（Executor 专用，不出 SSE）。"""
288	- type: Literal["sub_agent_dispatch"] = "sub_agent_dispatch"
289	- specs: list[SubAgentSpec]
290	-
291	-
292	-@dataclass(frozen=True)
293	-class SubAgentResult:
294	- """单个子 Agent 的执行结果（Executor 内部聚合用，不出 SSE）。"""
295	- entity_id: str
296	- status: Literal["done", "failed", "cancelled", "timeout"]
297	- content: str = "" # 子 Agent LLM 最终回答全量文本（来自 FinalAnswerChunkEvent.content；end 仅作结束标记，默认空）
298	- error: str = ""
299	- child_task_id: str = "" # 子 Agent A2A task_id，用于协议级取消和崩溃恢复
300	-
301	-
302	-@dataclass
303	-class WorkflowSpec:
304	- """单个并行工作流的派发规格（子 Agent 侧）。"""
305	- workflow_id: str # 唯一标识，作 SubTaskEvent.sub_task_path 工作流段（同父下唯一）；不传给 VA
306	- intent: str # VA 调用意图标识，对应 DelegateRequest.intent
307	- task_description: str # 自然语言任务描述，对应 DelegateRequest.task_description
308	- target_agent: str = "" # VA target_agent（agent_id），对应 DelegateRequest.target_agent
309	-
310	- def va_fields(self) -> dict:
311	- """提取 VA 调用所需字段，用于构造 DelegateRequest。"""
312	- return {
313	- "intent": self.intent,
314	- "task_description": self.task_description,
315	- "target_agent": self.target_agent or None,
316	- }
317	-
318	-
319	-class MultiDelegateRequest(BaseModel):
320	- """子 Agent yield 的并行工作流派发请求（Executor 专用，不出 SSE）。"""
321	- type: Literal["multi_delegate"] = "multi_delegate"
322	- workflows: list[WorkflowSpec]
323	-
324	-
325	-class SubTaskEvent(BaseModel):
326	- """级联盖章信封（北向唯一的子节点事件信封，进入 SSE 流、前端消费）。
327	-
328	- 子节点（子 Agent / 并行工作流）的原始帧原封不动放进 ``data``，外层只加
329	- ``sub_task`` 判别 + 节点绝对路径 + 节点类型；层级靠 ``sub_task_path`` 数组表达，
330	- wire 帧任意深度都单层（中间节点透传、不再套壳）。
331	-
332	- ``data`` 两种内容：
333	- - report 帧：子节点原始帧（agent 节点为 ``{type:"think_chunk",...}`` 等；
334	- workflow 节点为 VA 原生 ``{event:"message","data":<node>}``）。
335	- - 生命周期帧：
336	- ``{event:"node_start", entity_name\|intent:...}`` /
337	- ``{event:"node_end", status:"done\|failed\|cancelled\|timeout", content\|result\|error\|reason\|elapsed_ms:...}``
338	- """
339	- type: Literal["sub_task"] = "sub_task" # 北向 event 判别字段
340	- sub_task_path: list[str] # 绝对路径数组（root→生产者；段 = entity_id / workflow_id）
341	- node_kind: Literal["agent", "workflow"] # 节点类型，前端据此渲染（卡片 / 进度条）；每帧都带
342	- data: dict[str, Any] = Field(default_factory=dict) # 整帧原报文（原事件类型字段保留）
343	-
344	-
345	-# ════════════════════════════════════════════════════════════════════
346	-# 兼容事件（保留旧版 API，逐步废弃）
347	-# ════════════════════════════════════════════════════════════════════
348	-
349	-
350	-class ThoughtEvent(BaseModel):
351	- """[兼容] 旧版合并的 thought 事件，建议改用 Think* 系列。"""
352	- type: Literal["thought"] = "thought"
353	- content: str
354	-
355	-
356	-class AnswerEvent(BaseModel):
357	- """[兼容] 旧版合并的 answer 事件，建议改用 FinalAnswer* 系列。"""
358	- type: Literal["answer"] = "answer"
359	- content: str
360	- final: bool = False
361	-
362	-
363	-# ════════════════════════════════════════════════════════════════════
364	-# 统一 Union 类型
365	-# ════════════════════════════════════════════════════════════════════
366	-
367	-
368	-AgentEvent = Union[
369	- # 会话
370	- ConversationStartEvent,
371	- ConversationEndEvent,
372	- # 思考
373	- ThinkStartEvent,
374	- ThinkChunkEvent,
375	- ThinkEndEvent,
376	- # 规划
377	- TodoListStartEvent,
378	- TodoListItemEvent,
379	- TodoListEndEvent,
380	- # 任务
381	- TodoStartEvent,
382	- TodoStatusEvent,
383	- TodoEndEvent,
384	- # 工具
385	- ToolStartEvent,
386	- ToolStatusEvent,
387	- ToolEndEvent,
388	- # 执行轨迹
389	- PlanningExecutionProcessEvent,
390	- # 中断
391	- InterruptStartEvent,
392	- InterruptEndEvent,
393	- # 总结
394	- FinalAnswerStartEvent,
395	- SummaryEvent,
396	- FinalAnswerChunkEvent,
397	- FinalAnswerEndEvent,
398	- # Executor
399	- DelegateRequest,
400	- SubAgentDispatchRequest, # 主 Agent 侧并行子 Agent 派发（不出 SSE）
401	- MultiDelegateRequest, # 子 Agent 侧并行工作流派发（不出 SSE）
402	- SubTaskEvent, # 级联盖章信封（出 SSE，前端消费）
403	- # 兼容
404	- ThoughtEvent,
405	- AnswerEvent,
406	-]
407	-
408	-
409	-# 事件 type → 事件类（给 agent_adapter / user_router 用）
410	-EVENT_TYPE_MAP = {
411	- # 会话
412	- "conversation_start": ConversationStartEvent,
413	- "conversation_end": ConversationEndEvent,
414	- # 思考
415	- "think_start": ThinkStartEvent,
416	- "think_chunk": ThinkChunkEvent,
417	- "think_end": ThinkEndEvent,
418	- # 规划
419	- "todolist_start": TodoListStartEvent,
420	- "todolist_item": TodoListItemEvent,
421	- "todolist_end": TodoListEndEvent,
422	- # 任务
423	- "todo_start": TodoStartEvent,
424	- "todo_status": TodoStatusEvent,
425	- "todo_end": TodoEndEvent,
426	- # 工具
427	- "tool_start": ToolStartEvent,
428	- "tool_status": ToolStatusEvent,
429	- "tool_end": ToolEndEvent,
430	- # 执行轨迹
431	- "planning_execution_process": PlanningExecutionProcessEvent,
432	- # 中断
433	- "interrupt_start": InterruptStartEvent,
434	- "interrupt_end": InterruptEndEvent,
435	- # 总结
436	- "final_answer_start": FinalAnswerStartEvent,
437	- "summary": SummaryEvent,
438	- "final_answer_chunk": FinalAnswerChunkEvent,
439	- "final_answer_end": FinalAnswerEndEvent,
440	- # 级联盖章信封
441	- "sub_task": SubTaskEvent,
442	- # 兼容
443	- "thought": ThoughtEvent,
444	- "answer": AnswerEvent,
445	-}

Mapplications/a2a_service/common/redis_task_store.py+31-0

	@@ -51,3 +51,34 @@ class RedisTaskStore(TaskStore):
51	self, params: ListTasksRequest, context: ServerCallContext	51	self, params: ListTasksRequest, context: ServerCallContext
52	) -> ListTasksResponse:	52	) -> ListTasksResponse:
53	return ListTasksResponse()	53	return ListTasksResponse()
		54	+
		55	+
		56	+class ReadOnlyTaskStore(TaskStore):
		57	+ """只读 TaskStore 包装器：读/删/列表委托给底层 store，save 为空操作。
		58	+
		59	+ 用于 SDK DefaultRequestHandler，避免 SDK 的 TaskManager 对每个流式事件
		60	+ 都全量序列化 Task 写入 Redis（数千次 save 雪崩）。
		61	+ Task 状态持久化由应用层 TaskStateManager 统一管理：
		62	+ - create_task / update_task_status → save 到 Redis
		63	+ - finalize_completed → save COMPLETED 到 Redis
		64	+ SDK 仅在内存中维护 _current_task 处理事件流，artifacts 通过 SSE 实时透传上游，
		65	+ 无需持久化。
		66	+ """
		67	+
		68	+ def __init__(self, inner: TaskStore) -> None:
		69	+ self._inner = inner
		70	+
		71	+ async def save(self, task: Task, context: ServerCallContext) -> None:
		72	+ # 空操作：SDK 的事件级 save 不写 Redis，由应用层 TaskStateManager 管理状态持久化
		73	+ pass
		74	+
		75	+ async def get(self, task_id: str, context: ServerCallContext) -> Optional[Task]:
		76	+ return await self._inner.get(task_id, context)
		77	+
		78	+ async def delete(self, task_id: str, context: ServerCallContext) -> None:
		79	+ await self._inner.delete(task_id, context)
		80	+
		81	+ async def list(
		82	+ self, params: ListTasksRequest, context: ServerCallContext
		83	+ ) -> ListTasksResponse:
		84	+ return await self._inner.list(params, context)

Mapplications/a2a_service/common/response_wrapper.py+13-49

	@@ -80,8 +80,6 @@ def wrap_agent_event(
80	success: 顶层 ``success`` 字段。VA 上游报错或内部异常映射为 interrupt_start	80	success: 顶层 ``success`` 字段。VA 上游报错或内部异常映射为 interrupt_start
81	帧时设为 False，对齐 AgentEngine planning_agent 的失败帧形态。	81	帧时设为 False，对齐 AgentEngine planning_agent 的失败帧形态。
82	error: 顶层 ``error`` 字段。仅 success=False 时承载错误描述。	82	error: 顶层 ``error`` 字段。仅 success=False 时承载错误描述。
83	- display: 北向接口 v1.3 新增字段。true=前端拼接, false=前端停止拼接,
84	- None=不写入(向后兼容)。
85		83
86	Returns:	84	Returns:
87	可直接 json.dumps 的 dict，格式对齐规范文档 §4.4.3（agent event）。	85	可直接 json.dumps 的 dict，格式对齐规范文档 §4.4.3（agent event）。
	@@ -101,7 +99,6 @@ def wrap_agent_event(
101	"latency": "",	99	"latency": "",
102	"plugin": plugin,	100	"plugin": plugin,
103	}	101	}
104	- # 仅当 display 非 None 时才写入 custom_rsp_data，保持向后兼容
105	if display is not None:	102	if display is not None:
106	custom_rsp_data["display"] = display	103	custom_rsp_data["display"] = display
107		104
	@@ -164,7 +161,7 @@ def wrap_workflow_event(
164		161
165		162
166	# ════════════════════════════════════════════════════════════════════	163	# ════════════════════════════════════════════════════════════════════
167	-# sub_task 信封包装（级联盖章，对齐 TECH §3.2 / 北向消息定义文档 §2）	164	+# 错误 / 限流响应包装
168	# ════════════════════════════════════════════════════════════════════	165	# ════════════════════════════════════════════════════════════════════
169		166
170		167
	@@ -175,35 +172,26 @@ def _render_inner_custom_rsp_data(
175	conversation_id: str,	172	conversation_id: str,
176	elapsed: float,	173	elapsed: float,
177	) -> dict[str, Any]:	174	) -> dict[str, Any]:
178	- """把子节点原帧的 inner_meta 渲染成"标准单帧 custom_rsp_data"。
179	-
180	- 保证 ``sub_task.data`` 与该帧非并行单独渲染时逐字段一致（前端整段复用既有渲染器）：
181	- - ``agent``：复用 wrap_agent_event 的 custom_rsp_data（含 event/content/data/plugin…）；
182	- - ``workflow``：复用 wrap_workflow_event 的 custom_rsp_data（``{event, data}``）；
183	- - ``lifecycle``：node_start/node_end 为框架合成帧，按原样透传 ``{event, ...}``。
184	- """
185	kind = inner_meta.get("kind")	175	kind = inner_meta.get("kind")
186	if kind == "workflow":	176	if kind == "workflow":
187	return wrap_workflow_event(	177	return wrap_workflow_event(
188	- event_kind=inner_meta["type"],	178	+ event_kind=str(inner_meta.get("type") or "message"),
189	data=inner_meta.get("data") or {},	179	data=inner_meta.get("data") or {},
190	agent_id=agent_id,	180	agent_id=agent_id,
191	conversation_id=conversation_id,	181	conversation_id=conversation_id,
192	elapsed=elapsed,	182	elapsed=elapsed,
193	)["custom_rsp_data"]	183	)["custom_rsp_data"]
194	if kind == "lifecycle":	184	if kind == "lifecycle":
195	- # 框架合成的生命周期帧（node_start/node_end），已是 {event, ...} 形态，原样透传
196	return inner_meta.get("data") or {}	185	return inner_meta.get("data") or {}
197	- # 默认按 agent 帧渲染
198	return wrap_agent_event(	186	return wrap_agent_event(
199	- event_type=inner_meta.get("type", "thought"),	187	+ event_type=str(inner_meta.get("type") or "thought"),
200	- content=inner_meta.get("content", ""),	188	+ content=str(inner_meta.get("content") or ""),
201	data=inner_meta.get("data") or {},	189	data=inner_meta.get("data") or {},
202	agent_id=agent_id,	190	agent_id=agent_id,
203	conversation_id=conversation_id,	191	conversation_id=conversation_id,
204	elapsed=elapsed,	192	elapsed=elapsed,
205	- plugin=inner_meta.get("plugin", ""),	193	+ plugin=str(inner_meta.get("plugin") or ""),
206	- display=inner_meta.get("display"), # display（北向 v1.3）贯穿级联	194	+ display=inner_meta.get("display"),
207	)["custom_rsp_data"]	195	)["custom_rsp_data"]
208		196
209		197
	@@ -216,30 +204,6 @@ def wrap_sub_task_event(
216	conversation_id: str,	204	conversation_id: str,
217	elapsed: float,	205	elapsed: float,
218	) -> dict[str, Any]:	206	) -> dict[str, Any]:
219	- """包装级联盖章子节点帧（``custom_rsp_data.event == "sub_task"``）。
220	-
221	- 外层信封与 agent event 一致（success/agent_id/conversation_id/output/error/
222	- execution_time）；``custom_rsp_data`` 顶层带 ``event="sub_task"`` + ``sub_task_path``
223	- + ``node_kind``，``data`` 为子节点原帧渲染后的标准 custom_rsp_data（见
224	- ``_render_inner_custom_rsp_data``）。
225	-
226	- Args:
227	- sub_task_path: 节点绝对路径数组（root→生产者；段 = entity_id / workflow_id）。
228	- node_kind: ``"agent"`` 或 ``"workflow"``，前端据此渲染（卡片 / 进度条）。
229	- inner_meta: 子节点原帧的提取结果（``kind`` ∈ agent/workflow/lifecycle）。
230	- agent_id / conversation_id: 同 agent event。
231	- elapsed: 累计秒数。
232	-
233	- Returns:
234	- 可 json.dumps 的 dict，对齐北向消息定义文档 §2。子帧内部错误体现在
235	- ``data.node_end.status``，外层 ``success`` 恒为 true（=传输成功）。
236	- """
237	- inner = _render_inner_custom_rsp_data(
238	- inner_meta,
239	- agent_id=agent_id,
240	- conversation_id=conversation_id,
241	- elapsed=elapsed,
242	- )
243	return {	207	return {
244	"success": True,	208	"success": True,
245	"agent_id": agent_id,	209	"agent_id": agent_id,
	@@ -251,16 +215,16 @@ def wrap_sub_task_event(
251	"event": "sub_task",	215	"event": "sub_task",
252	"sub_task_path": sub_task_path,	216	"sub_task_path": sub_task_path,
253	"node_kind": node_kind,	217	"node_kind": node_kind,
254	- "data": inner,	218	+ "data": _render_inner_custom_rsp_data(
		219	+ inner_meta,
		220	+ agent_id=agent_id,
		221	+ conversation_id=conversation_id,
		222	+ elapsed=elapsed,
		223	+ ),
255	},	224	},
256	}	225	}
257		226
258		227
259	-# ════════════════════════════════════════════════════════════════════
260	-# 错误 / 限流响应包装
261	-# ════════════════════════════════════════════════════════════════════
262	-
263	-
264	def wrap_error(	228	def wrap_error(
265	*,	229	*,
266	agent_id: str,	230	agent_id: str,
	@@ -272,7 +236,7 @@ def wrap_error(
272	"""限流、内部错误等场景的简化响应。	236	"""限流、内部错误等场景的简化响应。
273		237
274	与成功帧相比：``success`` 为 False，带 ``error_code`` 和 ``error_msg``，	238	与成功帧相比：``success`` 为 False，带 ``error_code`` 和 ``error_msg``，
275	- 没有 ``custom_rsp_data``。当前限流分支（user_router.py:434-440）已经是	239	+ 没有 ``custom_rsp_data``。当前限流分支（api/dispatch.py）已经是
276	这种结构，保持兼容。	240	这种结构，保持兼容。
277		241
278	Args:	242	Args:

Dapplications/a2a_service/orchestrator/agent_adapter.py+0-155

	@@ -1,155 +0,0 @@
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	- 顶层主 Agent（final_answer_terminal=True，默认）：final_answer_end / AnswerEvent(final=True)
9	- → COMPLETED；其余事件 → TaskArtifactUpdateEvent(last_chunk=False)。
10	- 子 Agent（final_answer_terminal=False）：final_answer_end 也走 artifact，终态由
11	- Executor 在本轮自然结束时补发一次 COMPLETED——避免中途 final_answer_end 被 A2A server
12	- 当作终态而提前关流（详见 agent_event_to_a2a docstring）。
13	- artifact 的 parts 里携带：
14	- - Part(text=<content>) 可读文本（若有）
15	- - Part(data=<event fields>) 结构化数据，客户端按 data.type 分派
16	-
17	- DelegateRequest → None（Executor 直接处理）
18	-"""
19	-from __future__ import annotations
20	-
21	-import uuid
22	-from typing import Optional
23	-
24	-from a2a.types.a2a_pb2 import (
25	- Artifact,
26	- Message,
27	- Part,
28	- TaskArtifactUpdateEvent,
29	- TaskStatus,
30	- TaskStatusUpdateEvent,
31	- ROLE_AGENT,
32	- TASK_STATE_COMPLETED,
33	-)
34	-from google.protobuf.struct_pb2 import Struct, Value
35	-
36	-from common.events import (
37	- AgentEvent,
38	- # 会话
39	- ConversationStartEvent, ConversationEndEvent,
40	- # 思考
41	- ThinkStartEvent, ThinkChunkEvent, ThinkEndEvent,
42	- # 规划
43	- TodoListStartEvent, TodoListItemEvent, TodoListEndEvent,
44	- # 任务
45	- TodoStartEvent, TodoStatusEvent, TodoEndEvent,
46	- # 工具
47	- ToolStartEvent, ToolStatusEvent, ToolEndEvent,
48	- # 执行轨迹
49	- PlanningExecutionProcessEvent,
50	- # 中断
51	- InterruptStartEvent, InterruptEndEvent,
52	- # 总结
53	- FinalAnswerStartEvent, SummaryEvent, FinalAnswerChunkEvent, FinalAnswerEndEvent,
54	- # 并行子 Agent / 多工作流
55	- SubAgentDispatchRequest, MultiDelegateRequest, SubTaskEvent,
56	- # 兼容
57	- ThoughtEvent, AnswerEvent,
58	-)
59	-
60	-
61	-def _build_data_part(data: dict) -> Part:
62	- struct = Struct()
63	- # Struct 要求所有值必须是 JSON 可序列化的简单类型；pydantic model_dump 已经做过
64	- struct.update(data)
65	- value = Value()
66	- value.struct_value.CopyFrom(struct)
67	- part = Part()
68	- part.data.CopyFrom(value)
69	- return part
70	-
71	-
72	-def _build_artifact(text: str, data: dict) -> Artifact:
73	- parts = []
74	- if text:
75	- parts.append(Part(text=text))
76	- parts.append(_build_data_part(data))
77	- return Artifact(artifact_id=str(uuid.uuid4()), parts=parts)
78	-
79	-
80	-def _artifact_event(task_id: str, conv_id: str, artifact: Artifact) -> TaskArtifactUpdateEvent:
81	- return TaskArtifactUpdateEvent(
82	- task_id=task_id,
83	- context_id=conv_id,
84	- artifact=artifact,
85	- last_chunk=False,
86	- )
87	-
88	-
89	-def _completed(task_id: str, conv_id: str, content: str) -> TaskStatusUpdateEvent:
90	- part = Part(text=content)
91	- msg = Message(role=ROLE_AGENT, message_id=str(uuid.uuid4()), parts=[part])
92	- return TaskStatusUpdateEvent(
93	- task_id=task_id,
94	- context_id=conv_id,
95	- status=TaskStatus(state=TASK_STATE_COMPLETED, message=msg),
96	- )
97	-
98	-
99	-def agent_event_to_a2a(
100	- event: AgentEvent,
101	- task_id: str,
102	- conv_id: str,
103	- final_answer_terminal: bool = True,
104	-) -> Optional[TaskArtifactUpdateEvent \| TaskStatusUpdateEvent]:
105	- """AgentEvent → A2A 事件。DelegateRequest 返回 None 由 Executor 处理。
106	-
107	- ``final_answer_terminal``（默认 True，顶层主 Agent 路径）：把 FinalAnswerEndEvent /
108	- AnswerEvent(final=True) 映射为 COMPLETED 终态——顶层由 user_router 抽干队列到 close
109	- 才停，多个 COMPLETED 不会截断，历史前端报文序列不变。
110	- 子 Agent 路径须传 ``False``：子 Agent 经 A2A server 流式回传父侧，A2A server 一旦遇到
111	- COMPLETED 终态即判定任务结束、关闭流（见 a2a sdk active_task.py 终态处理）。规划型 Agent
112	- 在工具调用之间会发多段 final_answer_end（含首段开场白），若中途就发 COMPLETED，子 Agent
113	- 的真实 cascade 结果（call_versatile / call_multiversatile）将无法回传。传 False 时
114	- final_answer_end 降级为普通 artifact 透传，真正的终态由 Executor.run_agent 在本轮自然
115	- 结束时统一补发一次 COMPLETED。
116	- """
117	-
118	- # ── 终止态：TaskStatusUpdateEvent(COMPLETED)（仅顶层 final_answer_terminal=True）──
119	- if final_answer_terminal and isinstance(event, FinalAnswerEndEvent):
120	- return _completed(task_id, conv_id, event.content)
121	-
122	- # 兼容旧 AnswerEvent(final=True)
123	- if final_answer_terminal and isinstance(event, AnswerEvent) and event.final:
124	- return _completed(task_id, conv_id, event.content)
125	-
126	- # ── 派发请求：Executor 直接处理，不出 SSE ─────────────────────────
127	- if isinstance(event, (SubAgentDispatchRequest, MultiDelegateRequest)):
128	- return None
129	-
130	- # ── 级联盖章信封 SubTaskEvent → TaskArtifactUpdateEvent ───────────
131	- # data 带 type="sub_task" 判别：使上一跳 _drive_sub_agent 能识别为"已盖章"帧而透传，
132	- # 顶层 user_router._extract_event_meta 据此走 sub_task 信封分支。
133	- if isinstance(event, SubTaskEvent):
134	- data = event.model_dump() # {type:"sub_task", sub_task_path, node_kind, data:<原帧>}
135	- return _artifact_event(task_id, conv_id, _build_artifact("", data))
136	-
137	- # ── 其余事件统一 TaskArtifactUpdateEvent ──────────────────────────
138	- event_groups = (
139	- ConversationStartEvent, ConversationEndEvent,
140	- ThinkStartEvent, ThinkChunkEvent, ThinkEndEvent,
141	- TodoListStartEvent, TodoListItemEvent, TodoListEndEvent,
142	- TodoStartEvent, TodoStatusEvent, TodoEndEvent,
143	- ToolStartEvent, ToolStatusEvent, ToolEndEvent,
144	- PlanningExecutionProcessEvent,
145	- InterruptStartEvent, InterruptEndEvent,
146	- FinalAnswerStartEvent, SummaryEvent, FinalAnswerChunkEvent, FinalAnswerEndEvent,
147	- ThoughtEvent, AnswerEvent,
148	- )
149	- if isinstance(event, event_groups):
150	- data = event.model_dump()
151	- text = str(data.get("content", "") or "")
152	- return _artifact_event(task_id, conv_id, _build_artifact(text, data))
153	-
154	- # DelegateRequest 等由上层处理
155	- return None

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

The file is empty

Aapplications/a2a_service/orchestrator/config/route_config.yaml+37-0

@@ -0,0 +1,37 @@
	1	+handlers:
	2	+ local_agent: "orchestrator.handlers.local_agent_handler.LocalAgentHandler"
	3	+ remote_agent: "orchestrator.handlers.remote_agent_handler.RemoteAgentHandler"
	4	+ requester: "orchestrator.handlers.requester_handler.RequesterHandler"
	5	+
	6	+max_cascade_depth: 10
	7	+
	8	+default_profile:
	9	+ requester:
	10	+ suspended_states:
	11	+ - "INPUT_REQUIRED"
	12	+ - "AUTH_REQUIRED"
	13	+
	14	+ local_agent:
	15	+ delegate_types:
	16	+ - "delegate"
	17	+ - "sub_agent_dispatch"
	18	+ - "multi_delegate"
	19	+ default_remote_agent: "versatile_adapter"
	20	+
	21	+ remote_agent:
	22	+ terminal_frame_types:
	23	+ - "CONTROL_COMPLETED"
	24	+ - "CONTROL_FAILED"
	25	+ frame_type_map:
	26	+ COMPLETED: "CONTROL_COMPLETED"
	27	+ FAILED: "CONTROL_FAILED"
	28	+ INPUT_REQUIRED: "CONTROL_INPUT_REQUIRED"
	29	+ AUTH_REQUIRED: "CONTROL_AUTH_REQUIRED"
	30	+ SUBMITTED: "CONTROL_SUBMITTED"
	31	+ WORKING: "CONTROL_WORKING"
	32	+ CANCELED: "CONTROL_CANCELED"
	33	+ REJECTED: "CONTROL_REJECTED"
	34	+ ARTIFACT: "DATA"
	35	+ default_frame_type: "CONTROL_UNSPECIFIED"
	36	+
	37	+profiles: {}

Mapplications/a2a_service/orchestrator/executor.py+362-1732

Aapplications/a2a_service/orchestrator/handlers/__init__.py+9-0

@@ -0,0 +1,9 @@
	1	+from .requester_handler import RequesterHandler
	2	+from .local_agent_handler import LocalAgentHandler
	3	+from .remote_agent_handler import RemoteAgentHandler
	4	+
	5	+__all__ = [
	6	+ "RequesterHandler",
	7	+ "LocalAgentHandler",
	8	+ "RemoteAgentHandler",
	9	+]

Aapplications/a2a_service/orchestrator/handlers/local_agent_handler.py+75-0

@@ -0,0 +1,75 @@
	1	+from __future__ import annotations
	2	+
	3	+from typing import Any, Dict
	4	+
	5	+from loguru import logger
	6	+
	7	+from ..route.normalized_event import NormalizedEvent, RouteTarget
	8	+from ..state import TaskStateManager
	9	+
	10	+
	11	+class LocalAgentHandler:
	12	+ """Local Agent 目标处理器
	13	+
	14	+ 职责：
	15	+ - 处理 request 类型事件：首轮创建 Task + 启动 agent_stream
	16	+ - 处理 completed 类型事件：finalize_completed
	17	+
	18	+ 注意：
	19	+ - INPUT_REQUIRED 续轮逻辑已由 RequesterSourceStrategy 路由到 remote_agent 处理，
	20	+ 不再在此 handler 中判断
	21	+ - continue 事件由 RemoteAgentHandler 处理，不会路由到此处
	22	+ """
	23	+
	24	+ def __init__(self, state_manager: TaskStateManager, local_agent_name: str):
	25	+ self._state_manager = state_manager
	26	+ self._local_agent_name = local_agent_name
	27	+
	28	+ async def handle(
	29	+ self,
	30	+ event: NormalizedEvent,
	31	+ target: RouteTarget,
	32	+ context: Dict[str, Any],
	33	+ ) -> None:
	34	+ event_type = event.type
	35	+ task_id = context.get("task_id", "")
	36	+ conv_id = context.get("conv_id", "")
	37	+ call_context = context.get("call_context")
	38	+ current_task = context.get("current_task")
	39	+ executor = context.get("executor")
	40	+ turn_ctx = context.get("turn_ctx")
	41	+ query = context.get("query", "")
	42	+ original_body = context.get("original_body", {})
	43	+ step_counter = context.get("step_counter")
	44	+
	45	+ if event_type == "request":
	46	+ if current_task is None:
	47	+ await self._state_manager.create_task(
	48	+ task_id, conv_id, call_context=call_context,
	49	+ source_agent=self._local_agent_name,
	50	+ )
	51	+ logger.debug(
	52	+ f"[LocalAgentHandler] 创建 Task：task={task_id}, conv={conv_id}, source_agent={self._local_agent_name}"
	53	+ )
	54	+ if executor is not None and turn_ctx is not None:
	55	+ await executor.run_agent(
	56	+ turn_ctx,
	57	+ query=query,
	58	+ original_body=original_body,
	59	+ cascade_result=None,
	60	+ step_counter=step_counter or [0],
	61	+ )
	62	+ return
	63	+
	64	+ # NOTE: completed 分支当前为预留逻辑，尚未有业务路径触发。
	65	+ # 触发条件：RemoteAgentHandler 将 VA 返回的终态事件（如 COMPLETED）
	66	+ # 归一化为 NormalizedEvent(source=remote_agent, type=completed) 后走 dispatch 路由，
	67	+ # 由 RemoteAgentSourceStrategy 将 CONTROL_COMPLETED 帧路由到 local_agent 目标。
	68	+ # 目前 VA 事件仍由 RemoteAgentHandler 直接 enqueue_event，未走 dispatch，
	69	+ # 因此该分支暂不可达。待 VA 事件统一走 dispatch 路由后激活。
	70	+ if event_type == "completed":
	71	+ await self._state_manager.finalize_completed(task_id, call_context)
	72	+ logger.debug(
	73	+ f"[LocalAgentHandler] Task COMPLETED: task={task_id}, conv={conv_id}"
	74	+ )
	75	+ return

Aapplications/a2a_service/orchestrator/handlers/remote_agent_handler.py+1642-0

Aapplications/a2a_service/orchestrator/handlers/requester_handler.py+182-0

@@ -0,0 +1,182 @@
	1	+from __future__ import annotations
	2	+
	3	+from typing import Any, Dict
	4	+
	5	+from a2a.types.a2a_pb2 import (
	6	+ TaskStatus,
	7	+ TaskStatusUpdateEvent,
	8	+ TASK_STATE_INPUT_REQUIRED,
	9	+)
	10	+from loguru import logger
	11	+
	12	+from channels.dict_to_a2a import dict_to_a2a
	13	+from ..route.normalized_event import NormalizedEvent, RouteTarget
	14	+from ..state import InputRequiredState, TaskStateManager
	15	+
	16	+
	17	+def _raw_data(raw_event: Any) -> dict[str, Any]:
	18	+ if isinstance(raw_event, dict):
	19	+ data = raw_event.get("data", {})
	20	+ return data if isinstance(data, dict) else {}
	21	+ return {}
	22	+
	23	+
	24	+def _agent_dict_to_a2a(raw_event: Any, task_id: str, conv_id: str):
	25	+ if not isinstance(raw_event, dict):
	26	+ return None
	27	+
	28	+ event_type = raw_event.get("type")
	29	+ data = raw_event.get("data") or {}
	30	+ if not isinstance(data, dict):
	31	+ data = {}
	32	+
	33	+ if event_type == "delegate":
	34	+ return None
	35	+
	36	+ return dict_to_a2a({"type": event_type, "data": data}, task_id, conv_id)
	37	+
	38	+
	39	+class RequesterHandler:
	40	+ """Requester 目标处理器（转发事件到请求端，请求端可以是 Channel 或请求端 Agent）
	41	+
	42	+ 处理来自 local_agent 的归一化事件：
	43	+ - tool_start：step 边界发射 + planning_execution_process + 原始事件 a2a 入队
	44	+ - 其他事件（think_start / text_chunk / tool_end 等）：raw_event a2a 转换 + 入队
	45	+ - input_required：挂起任务状态
	46	+ - failed：标记任务失败
	47	+ """
	48	+
	49	+ def __init__(self, state_manager: TaskStateManager):
	50	+ self._state_manager = state_manager
	51	+
	52	+ async def handle(
	53	+ self,
	54	+ event: NormalizedEvent,
	55	+ target: RouteTarget,
	56	+ context: Dict[str, Any],
	57	+ ) -> None:
	58	+ event_type = event.type
	59	+ event_data = event.data
	60	+ event_queue = context.get("event_queue")
	61	+ task_id = context.get("task_id", "")
	62	+ conv_id = context.get("conv_id", "")
	63	+ call_context = context.get("call_context")
	64	+ step_counter = context.get("step_counter")
	65	+
	66	+ if event_type == "tool_start":
	67	+ await self._handle_tool_start(
	68	+ event_data, task_id, conv_id, event_queue, step_counter,
	69	+ )
	70	+ return
	71	+
	72	+ if event_type == "input_required":
	73	+ await self._handle_input_required(
	74	+ event_data, task_id, conv_id, call_context, event_queue
	75	+ )
	76	+ return
	77	+
	78	+ if event_type == "failed":
	79	+ await self._handle_failed(event_data, task_id, conv_id, call_context)
	80	+ return
	81	+
	82	+ await self._handle_raw_event(event_data, task_id, conv_id, event_queue)
	83	+
	84	+ async def _handle_tool_start(
	85	+ self,
	86	+ event_data: Dict[str, Any],
	87	+ task_id: str,
	88	+ conv_id: str,
	89	+ event_queue: Any,
	90	+ step_counter: Any,
	91	+ ) -> None:
	92	+ raw_event = event_data.get("raw_event")
	93	+ if raw_event is None:
	94	+ return
	95	+
	96	+ data = _raw_data(raw_event)
	97	+ plugin = str(data.get("plugin") or "")
	98	+ desc = str(data.get("content") or plugin)
	99	+
	100	+ if step_counter is not None:
	101	+ step_counter[0] += 1
	102	+
	103	+ planning_content = (
	104	+ f"[执行轨迹] 正在执行步骤{step_counter[0]}: {desc} "
	105	+ f"(tool={plugin})"
	106	+ )
	107	+ planning = {
	108	+ "type": "planning_execution_process",
	109	+ "data": {"content": planning_content},
	110	+ }
	111	+ planning_a2a = dict_to_a2a(planning, task_id, conv_id)
	112	+ if planning_a2a is not None and event_queue is not None:
	113	+ await event_queue.enqueue_event(planning_a2a)
	114	+
	115	+ logger.info(
	116	+ f"[RequesterHandler] step 边界: 步骤{step_counter[0]} "
	117	+ f"(tool={plugin}, desc={desc!r:.60})"
	118	+ )
	119	+
	120	+ a2a_event = _agent_dict_to_a2a(raw_event, task_id, conv_id)
	121	+ if a2a_event is not None and event_queue is not None:
	122	+ await event_queue.enqueue_event(a2a_event)
	123	+
	124	+ async def _handle_raw_event(
	125	+ self,
	126	+ event_data: Dict[str, Any],
	127	+ task_id: str,
	128	+ conv_id: str,
	129	+ event_queue: Any,
	130	+ ) -> None:
	131	+ raw_event = event_data.get("raw_event")
	132	+ if raw_event is None:
	133	+ return
	134	+ a2a_event = _agent_dict_to_a2a(raw_event, task_id, conv_id)
	135	+ if a2a_event is not None and event_queue is not None:
	136	+ await event_queue.enqueue_event(a2a_event)
	137	+
	138	+ async def _handle_input_required(
	139	+ self,
	140	+ event_data: Dict[str, Any],
	141	+ task_id: str,
	142	+ conv_id: str,
	143	+ call_context: Any,
	144	+ event_queue: Any,
	145	+ ) -> None:
	146	+ remote_task_id = event_data.get("remote_task_id", "")
	147	+
	148	+ await self._state_manager.save_input_required(
	149	+ InputRequiredState(
	150	+ task_id=task_id,
	151	+ call_context=call_context,
	152	+ remote_task_id=remote_task_id,
	153	+ workflow_id=event_data.get("workflow_id"),
	154	+ )
	155	+ )
	156	+
	157	+ if event_queue is not None:
	158	+ await event_queue.enqueue_event(
	159	+ TaskStatusUpdateEvent(
	160	+ task_id=task_id,
	161	+ context_id=conv_id,
	162	+ status=TaskStatus(state=TASK_STATE_INPUT_REQUIRED),
	163	+ )
	164	+ )
	165	+ logger.info(
	166	+ f"[RequesterHandler] INPUT_REQUIRED: task={task_id}, "
	167	+ f"conv={conv_id}, remote_task_id={remote_task_id}"
	168	+ )
	169	+
	170	+ async def _handle_failed(
	171	+ self,
	172	+ event_data: Dict[str, Any],
	173	+ task_id: str,
	174	+ conv_id: str,
	175	+ call_context: Any,
	176	+ ) -> None:
	177	+ error = event_data.get("error", "Agent failed")
	178	+ await self._state_manager.finalize_failed(task_id, call_context, error_text=error)
	179	+ logger.warning(
	180	+ f"[RequesterHandler] FAILED: task={task_id}, conv={conv_id}, "
	181	+ f"error={error!r:.80}"
	182	+ )

Aapplications/a2a_service/orchestrator/route/__init__.py+17-0

@@ -0,0 +1,17 @@
	1	+from .route_profiles import (
	2	+ RequesterSourceProfile,
	3	+ LocalAgentSourceProfile,
	4	+ RemoteAgentSourceProfile,
	5	+ SourceRouteProfile,
	6	+ RouteConfig,
	7	+ RouteConfigLoader,
	8	+)
	9	+from .route_strategies import (
	10	+ RouteStrategy,
	11	+ RequesterSourceStrategy,
	12	+ LocalAgentSourceStrategy,
	13	+ RemoteAgentSourceStrategy,
	14	+)
	15	+from .route_dispatcher import RouteDispatcher
	16	+from .normalized_event import NormalizedEvent, RouteContext, RouteTarget
	17	+from .handler_registry import HandlerRegistry

Aapplications/a2a_service/orchestrator/route/handler_registry.py+71-0

@@ -0,0 +1,71 @@
	1	+from __future__ import annotations
	2	+
	3	+import inspect
	4	+import importlib
	5	+from typing import Dict, Type
	6	+
	7	+from loguru import logger
	8	+
	9	+
	10	+class HandlerRegistry:
	11	+ """处理器注册表（插件化加载核心）
	12	+
	13	+ 支持从配置动态加载处理器插件，实现核心框架与业务逻辑解耦。
	14	+ """
	15	+
	16	+ def __init__(self):
	17	+ self._handler_classes: Dict[str, Type] = {}
	18	+
	19	+ def load_handlers(self, handler_config: Dict[str, str], **kwargs) -> Dict[str, object]:
	20	+ """从配置动态加载处理器类
	21	+
	22	+ Args:
	23	+ handler_config: 处理器配置 {target_type: class_path}
	24	+ **kwargs: 传递给处理器构造函数的参数（自动过滤非匹配参数）
	25	+
	26	+ Returns:
	27	+ 处理器实例映射 {target_type: handler_instance.handle}
	28	+ """
	29	+ handlers: Dict[str, object] = {}
	30	+
	31	+ for target_type, class_path in handler_config.items():
	32	+ try:
	33	+ module_path, class_name = class_path.rsplit(".", 1)
	34	+ module = importlib.import_module(module_path)
	35	+ handler_class = getattr(module, class_name)
	36	+
	37	+ filtered_kwargs = self._filter_kwargs(handler_class, kwargs)
	38	+ handler_instance = handler_class(**filtered_kwargs)
	39	+ handlers[target_type] = handler_instance
	40	+ logger.debug(
	41	+ f"[HandlerRegistry] 加载处理器: {target_type} -> {class_path}"
	42	+ )
	43	+ except Exception as e:
	44	+ logger.error(
	45	+ f"[HandlerRegistry] 加载处理器失败: {target_type} -> {class_path}, "
	46	+ f"error={e}"
	47	+ )
	48	+
	49	+ return handlers
	50	+
	51	+ @staticmethod
	52	+ def _filter_kwargs(handler_class: Type, kwargs: Dict) -> Dict:
	53	+ """根据处理器构造函数的签名过滤参数，只传递构造函数接受的参数"""
	54	+ try:
	55	+ sig = inspect.signature(handler_class.__init__)
	56	+ params = sig.parameters
	57	+
	58	+ if any(
	59	+ p.kind == inspect.Parameter.VAR_KEYWORD
	60	+ for p in params.values()
	61	+ ):
	62	+ return kwargs
	63	+
	64	+ accepted = set(params.keys()) - {"self"}
	65	+ return {k: v for k, v in kwargs.items() if k in accepted}
	66	+ except Exception:
	67	+ return kwargs
	68	+
	69	+ def register_handler_class(self, target_type: str, handler_class: Type):
	70	+ """注册处理器类（编程式注册，用于扩展）"""
	71	+ self._handler_classes[target_type] = handler_class

Aapplications/a2a_service/orchestrator/route/normalized_event.py+27-0

@@ -0,0 +1,27 @@
	1	+from __future__ import annotations
	2	+
	3	+from typing import Any, Dict, Optional
	4	+
	5	+from pydantic import BaseModel, ConfigDict, Field
	6	+
	7	+
	8	+class NormalizedEvent(BaseModel):
	9	+ type: str = Field(description="事件类型")
	10	+ data: Dict[str, Any] = Field(default_factory=dict, description="事件数据")
	11	+ metadata: Dict[str, Any] = Field(default_factory=dict, description="元数据（含 source、frame_type 等）")
	12	+
	13	+
	14	+class RouteContext(BaseModel):
	15	+ task_id: str = Field(default="", description="当前任务 ID")
	16	+ current_task: Optional[Any] = Field(default=None, description="当前 Task 对象")
	17	+ conv_id: str = Field(default="", description="会话 ID（conversation_id）")
	18	+ root_task_id: str = Field(default="", description="根任务 ID（首轮由 api.dispatch 通过 conv_id→task_id 映射查得）")
	19	+ agent_key: str = Field(default="", description="当前节点 Agent 名称（即 source_agent）")
	20	+ is_specify_task: bool = Field(default=True, description="是否指定了 task_id（False 表示老版本未传入 task_id，走兼容逻辑）")
	21	+
	22	+ model_config = ConfigDict(arbitrary_types_allowed=True)
	23	+
	24	+
	25	+class RouteTarget(BaseModel):
	26	+ type: str = Field(description="路由目标类型: local_agent / remote_agent / channel")
	27	+ agent_key: str = Field(default="", description="目标 Agent 标识（type=remote_agent 时使用）")

Aapplications/a2a_service/orchestrator/route/route_dispatcher.py+148-0

@@ -0,0 +1,148 @@
	1	+from __future__ import annotations
	2	+
	3	+from typing import Any, Callable, Dict, Optional, Set
	4	+
	5	+from loguru import logger
	6	+
	7	+from .handler_registry import HandlerRegistry
	8	+from .normalized_event import NormalizedEvent, RouteContext, RouteTarget
	9	+from .route_profiles import RouteConfig, RouteConfigLoader, SourceRouteProfile
	10	+from .route_strategies import (
	11	+ LocalAgentSourceStrategy,
	12	+ RemoteAgentSourceStrategy,
	13	+ RequesterSourceStrategy,
	14	+ RouteStrategy,
	15	+)
	16	+from ..state.task_state_manager import TaskStateManager
	17	+
	18	+
	19	+class RouteDispatcher:
	20	+ def __init__(
	21	+ self,
	22	+ state_mgr: TaskStateManager,
	23	+ config: Optional[RouteConfig] = None,
	24	+ config_path: Optional[str] = None,
	25	+ local_agent_names: Optional[list[str]] = None,
	26	+ ):
	27	+ self._state_mgr = state_mgr
	28	+ self._strategy_cache: Dict[str, Dict[str, RouteStrategy]] = {}
	29	+ self._handlers: Dict[str, Dict[Optional[str], Callable]] = {}
	30	+ self._handler_instances: Dict[str, object] = {}
	31	+ self._local_agent_keys: Set[str] = set(local_agent_names or [])
	32	+
	33	+ if config_path:
	34	+ self._config = RouteConfigLoader.load(config_path)
	35	+ else:
	36	+ self._config: RouteConfig = config or RouteConfig()
	37	+
	38	+ self._default_profile = self._config.default_profile
	39	+ self._profiles: Dict[str, SourceRouteProfile] = self._config.profiles
	40	+
	41	+ @property
	42	+ def config(self) -> RouteConfig:
	43	+ return self._config
	44	+
	45	+ def load_config(self, config_path: str) -> Dict[str, str]:
	46	+ """从 yaml 文件加载路由配置
	47	+
	48	+ Returns:
	49	+ 处理器配置字典 {target_type: handler_class_path}
	50	+ """
	51	+ self._config = RouteConfigLoader.load(config_path)
	52	+ self._default_profile = self._config.default_profile
	53	+ self._profiles = self._config.profiles
	54	+ self._strategy_cache.clear()
	55	+ self._ensure_strategies("")
	56	+ return self._config.handlers
	57	+
	58	+ def register_handlers_from_config(self, **kwargs) -> None:
	59	+ """从配置中的 handlers 字段动态加载并注册处理器插件
	60	+
	61	+ Args:
	62	+ **kwargs: 传递给处理器构造函数的参数（如 state_manager）
	63	+ """
	64	+ if not self._config.handlers:
	65	+ return
	66	+
	67	+ registry = HandlerRegistry()
	68	+ handlers = registry.load_handlers(self._config.handlers, **kwargs)
	69	+
	70	+ for target_type, handler_instance in handlers.items():
	71	+ self._handler_instances[target_type] = handler_instance
	72	+ self.register_handler(target_type, handler_instance.handle)
	73	+ logger.debug(
	74	+ f"[RouteDispatcher] 注册处理器: {target_type}"
	75	+ )
	76	+
	77	+ def get_handler_instance(self, target_type: str) -> object \| None:
	78	+ return self._handler_instances.get(target_type)
	79	+
	80	+ def get_profile(self, agent_key: str) -> SourceRouteProfile:
	81	+ return self._profiles.get(agent_key, self._default_profile)
	82	+
	83	+ def _ensure_strategies(self, agent_key: str) -> Dict[str, RouteStrategy]:
	84	+ if agent_key in self._strategy_cache:
	85	+ return self._strategy_cache[agent_key]
	86	+
	87	+ profile = self.get_profile(agent_key)
	88	+ strategies = {
	89	+ "requester": RequesterSourceStrategy(
	90	+ self._state_mgr, profile.requester, self._local_agent_keys, self._config.max_cascade_depth
	91	+ ),
	92	+ "local_agent": LocalAgentSourceStrategy(profile.local_agent),
	93	+ "remote_agent": RemoteAgentSourceStrategy(profile.remote_agent),
	94	+ }
	95	+ self._strategy_cache[agent_key] = strategies
	96	+ return strategies
	97	+
	98	+ def register_handler(
	99	+ self, target_type: str, handler: Callable, source: Optional[str] = None
	100	+ ):
	101	+ if target_type not in self._handlers:
	102	+ self._handlers[target_type] = {}
	103	+ self._handlers[target_type][source] = handler
	104	+
	105	+ async def route(self, event: NormalizedEvent, context: RouteContext) -> RouteTarget:
	106	+ source = event.metadata.get("source", "requester")
	107	+ agent_key = context.agent_key or ""
	108	+ strategies = self._ensure_strategies(agent_key)
	109	+
	110	+ strategy = strategies.get(source)
	111	+ if not strategy:
	112	+ raise ValueError(f"Unknown source direction: {source}")
	113	+
	114	+ target = await strategy.route(event, context)
	115	+
	116	+ # 关键日志：记录路由决策结果（源 -> 目标）
	117	+ logger.info(
	118	+ f"[RouteDispatcher] 路由决策: event_type={event.type}, "
	119	+ f"source={source} -> target={target.type}, agent_key={target.agent_key}"
	120	+ )
	121	+
	122	+ return target
	123	+
	124	+ async def dispatch(self, event: NormalizedEvent, context: Dict[str, Any]) -> Any:
	125	+ route_context = RouteContext(
	126	+ task_id=context.get("task_id", ""),
	127	+ current_task=context.get("current_task"),
	128	+ conv_id=context.get("conv_id", ""),
	129	+ root_task_id=context.get("root_task_id", ""),
	130	+ agent_key=context.get("agent_key", ""),
	131	+ is_specify_task=context.get("is_specify_task", True),
	132	+ )
	133	+ target = await self.route(event, route_context)
	134	+
	135	+ target_handlers = self._handlers.get(target.type)
	136	+ if not target_handlers:
	137	+ raise ValueError(f"No handler registered for target type: {target.type}")
	138	+
	139	+ event_source = event.metadata.get("source")
	140	+ handler = target_handlers.get(event_source) or target_handlers.get(None)
	141	+
	142	+ if not handler:
	143	+ raise ValueError(
	144	+ f"No handler registered for target type '{target.type}' "
	145	+ f"with source '{event_source}'."
	146	+ )
	147	+
	148	+ return await handler(event, target, context)

Aapplications/a2a_service/orchestrator/route/route_profiles.py+95-0

@@ -0,0 +1,95 @@
	1	+from __future__ import annotations
	2	+
	3	+from pathlib import Path
	4	+from typing import Any, Dict, List, Optional
	5	+
	6	+from pydantic import BaseModel, Field
	7	+
	8	+
	9	+class RequesterSourceProfile(BaseModel):
	10	+ suspended_states: List[str] = Field(
	11	+ default_factory=lambda: ["INPUT_REQUIRED", "AUTH_REQUIRED"],
	12	+ description="挂起状态列表（续轮判断条件）",
	13	+ )
	14	+
	15	+
	16	+class LocalAgentSourceProfile(BaseModel):
	17	+ delegate_types: List[str] = Field(
	18	+ default_factory=lambda: ["delegate", "sub_agent_dispatch", "multi_delegate"],
	19	+ description="委托事件类型集合（event.type 在此集合中 → 路由到 Remote Agent）",
	20	+ )
	21	+ default_remote_agent: str = Field(
	22	+ default="versatile_adapter",
	23	+ description="默认远程 Agent 标识（delegate 事件未指定 agent_key 时使用）",
	24	+ )
	25	+
	26	+
	27	+class RemoteAgentSourceProfile(BaseModel):
	28	+ terminal_frame_types: List[str] = Field(
	29	+ default_factory=lambda: ["CONTROL_COMPLETED", "CONTROL_FAILED"],
	30	+ description="终态帧类型集合（frame_type 在此集合中 → 路由到 Local Agent）",
	31	+ )
	32	+ frame_type_map: Dict[str, str] = Field(
	33	+ default_factory=lambda: {
	34	+ "COMPLETED": "CONTROL_COMPLETED",
	35	+ "FAILED": "CONTROL_FAILED",
	36	+ "INPUT_REQUIRED": "CONTROL_INPUT_REQUIRED",
	37	+ "AUTH_REQUIRED": "CONTROL_AUTH_REQUIRED",
	38	+ "SUBMITTED": "CONTROL_SUBMITTED",
	39	+ "WORKING": "CONTROL_WORKING",
	40	+ "CANCELED": "CONTROL_CANCELED",
	41	+ "REJECTED": "CONTROL_REJECTED",
	42	+ "ARTIFACT": "DATA",
	43	+ },
	44	+ description="A2A 事件状态 → frame_type 的映射表",
	45	+ )
	46	+ default_frame_type: str = Field(
	47	+ default="CONTROL_UNSPECIFIED",
	48	+ description="默认帧类型（映射表中未匹配时使用）",
	49	+ )
	50	+
	51	+
	52	+class SourceRouteProfile(BaseModel):
	53	+ requester: RequesterSourceProfile = Field(default_factory=RequesterSourceProfile)
	54	+ local_agent: LocalAgentSourceProfile = Field(default_factory=LocalAgentSourceProfile)
	55	+ remote_agent: RemoteAgentSourceProfile = Field(default_factory=RemoteAgentSourceProfile)
	56	+
	57	+
	58	+class RouteConfig(BaseModel):
	59	+ handlers: Dict[str, str] = Field(
	60	+ default_factory=dict,
	61	+ description="处理器插件映射（目标类型 → 处理器类路径）",
	62	+ )
	63	+ profiles: Dict[str, SourceRouteProfile] = Field(
	64	+ default_factory=dict,
	65	+ description="路由策略配置（Agent 名称 → SourceRouteProfile），支持多 Agent 独立配置",
	66	+ )
	67	+ default_profile: SourceRouteProfile = Field(
	68	+ default_factory=SourceRouteProfile,
	69	+ description="默认路由策略配置（未匹配到 Agent 专用配置时使用）",
	70	+ )
	71	+ max_cascade_depth: int = Field(
	72	+ default=10,
	73	+ description="级联查找最大深度限制，防止循环嵌套导致无限递归（默认10）",
	74	+ )
	75	+
	76	+
	77	+class RouteConfigLoader:
	78	+ """路由配置加载器：从 YAML 文件加载配置，支持多 Agent 配置合并"""
	79	+
	80	+ @staticmethod
	81	+ def load(config_path: str) -> RouteConfig:
	82	+ path = Path(config_path)
	83	+ if not path.exists():
	84	+ return RouteConfig()
	85	+
	86	+ import yaml
	87	+
	88	+ with open(path, "r", encoding="utf-8") as f:
	89	+ raw = yaml.safe_load(f) or {}
	90	+
	91	+ return RouteConfig(**raw)
	92	+
	93	+ @staticmethod
	94	+ def resolve_profile(config: RouteConfig, agent_key: str) -> SourceRouteProfile:
	95	+ return config.profiles.get(agent_key, config.default_profile)

Aapplications/a2a_service/orchestrator/route/route_strategies.py+174-0

@@ -0,0 +1,174 @@
	1	+from __future__ import annotations
	2	+
	3	+from abc import ABC, abstractmethod
	4	+from typing import Dict, List, Optional, Set
	5	+
	6	+from loguru import logger
	7	+
	8	+from .normalized_event import NormalizedEvent, RouteContext, RouteTarget
	9	+from .route_profiles import (
	10	+ RequesterSourceProfile,
	11	+ LocalAgentSourceProfile,
	12	+ RemoteAgentSourceProfile,
	13	+)
	14	+from ..state.task_state_manager import (
	15	+ TaskStateManager,
	16	+ META_KEY_SOURCE_AGENT,
	17	+ META_KEY_SUB_TASKS,
	18	+ META_KEY_REMOTE_TASK_ID,
	19	+)
	20	+
	21	+
	22	+class RouteStrategy(ABC):
	23	+ @abstractmethod
	24	+ async def route(self, event: NormalizedEvent, context: RouteContext) -> RouteTarget:
	25	+ ...
	26	+
	27	+
	28	+class RequesterSourceStrategy(RouteStrategy):
	29	+ def __init__(
	30	+ self,
	31	+ state_mgr: TaskStateManager,
	32	+ profile: RequesterSourceProfile,
	33	+ local_agent_keys: Optional[Set[str]] = None,
	34	+ max_cascade_depth: int = 10,
	35	+ ):
	36	+ self._state_mgr = state_mgr
	37	+ self._profile = profile
	38	+ self._suspended_states: Set[str] = set(profile.suspended_states)
	39	+ self._local_agent_keys: Set[str] = local_agent_keys or set()
	40	+ self._max_cascade_depth: int = max_cascade_depth
	41	+
	42	+ async def route(self, event: NormalizedEvent, context: RouteContext) -> RouteTarget:
	43	+ target_task_id = context.task_id or context.root_task_id
	44	+ task = await self._state_mgr.get_task(target_task_id)
	45	+
	46	+ # 临时调试打印
	47	+ remote_task_id_debug = task.get("metadata", {}).get(META_KEY_REMOTE_TASK_ID, "") if task else ""
	48	+ status_state_debug = task.get("status_state", "") if task else "None"
	49	+ logger.info(
	50	+ f"[RequesterSourceStrategy] route debug: "
	51	+ f"task_id={context.task_id}, root_task_id={context.root_task_id}, "
	52	+ f"target_task_id={target_task_id}, remote_task_id={remote_task_id_debug}, "
	53	+ f"status_state={status_state_debug}, is_specify_task={context.is_specify_task}"
	54	+ )
	55	+
	56	+ if task is None or task.get("status_state") not in self._suspended_states:
	57	+ return RouteTarget(type="local_agent")
	58	+
	59	+ # 老版本兼容：未指定 task_id 时，仅通过 remote_task_id 判断是否路由到远程代理
	60	+ if not context.is_specify_task:
	61	+ remote_task_id = task.get("metadata", {}).get(META_KEY_REMOTE_TASK_ID, "")
	62	+ if remote_task_id:
	63	+ remote_task = await self._state_mgr.get_task(remote_task_id)
	64	+ if remote_task:
	65	+ remote_source = remote_task.get("metadata", {}).get(META_KEY_SOURCE_AGENT, "")
	66	+ return RouteTarget(type="remote_agent", agent_key=remote_source)
	67	+ return RouteTarget(type="local_agent")
	68	+
	69	+ # 新版本：指定了 task_id，走级联路由逻辑
	70	+ source_agent = task.get("metadata", {}).get(META_KEY_SOURCE_AGENT, "")
	71	+ if source_agent in self._local_agent_keys:
	72	+ return RouteTarget(type="local_agent")
	73	+
	74	+ route_path = await self._resolve_route_path(context)
	75	+ next_hop = self._determine_next_hop(route_path)
	76	+ return RouteTarget(type="remote_agent", agent_key=next_hop)
	77	+
	78	+ async def _resolve_route_path(self, context: RouteContext) -> List[dict]:
	79	+ root_task_id = context.root_task_id
	80	+ root_task = await self._state_mgr.get_task(root_task_id)
	81	+ if not root_task:
	82	+ return []
	83	+
	84	+ route_path = [
	85	+ {
	86	+ "task_id": root_task_id,
	87	+ "source_agent": root_task.get("metadata", {}).get(META_KEY_SOURCE_AGENT, ""),
	88	+ }
	89	+ ]
	90	+ target_task_id = context.task_id
	91	+
	92	+ if not target_task_id or target_task_id == root_task_id:
	93	+ return route_path
	94	+
	95	+ found = await self._bfs_find(root_task, target_task_id, route_path, depth=1)
	96	+ return found
	97	+
	98	+ async def _bfs_find(
	99	+ self,
	100	+ parent_task: dict,
	101	+ target_task_id: str,
	102	+ path: List[dict],
	103	+ depth: int = 1,
	104	+ ) -> List[dict]:
	105	+ if depth > self._max_cascade_depth:
	106	+ return path
	107	+
	108	+ sub_task_ids = parent_task.get("metadata", {}).get(META_KEY_SUB_TASKS, [])
	109	+ if not sub_task_ids:
	110	+ return path
	111	+
	112	+ for sub_id in sub_task_ids:
	113	+ sub_task = await self._state_mgr.get_task(sub_id)
	114	+ if not sub_task:
	115	+ continue
	116	+ sub_path = path + [
	117	+ {
	118	+ "task_id": sub_id,
	119	+ "source_agent": sub_task.get("metadata", {}).get(META_KEY_SOURCE_AGENT, ""),
	120	+ }
	121	+ ]
	122	+ if sub_id == target_task_id:
	123	+ return sub_path
	124	+ result = await self._bfs_find(sub_task, target_task_id, sub_path, depth=depth + 1)
	125	+ if any(node["task_id"] == target_task_id for node in result):
	126	+ return result
	127	+ return path
	128	+
	129	+ def _determine_next_hop(self, route_path: List[dict]) -> str:
	130	+ if not route_path:
	131	+ return ""
	132	+ for i in range(len(route_path) - 1, -1, -1):
	133	+ if route_path[i]["source_agent"] in self._local_agent_keys and i + 1 < len(route_path):
	134	+ return route_path[i + 1]["source_agent"]
	135	+ return route_path[-1].get("source_agent", "")
	136	+
	137	+
	138	+class LocalAgentSourceStrategy(RouteStrategy):
	139	+ def __init__(self, profile: LocalAgentSourceProfile):
	140	+ self._profile = profile
	141	+ self._delegate_types: Set[str] = set(profile.delegate_types)
	142	+
	143	+ async def route(self, event: NormalizedEvent, context: RouteContext) -> RouteTarget:
	144	+ if event.type in self._delegate_types:
	145	+ agent_key = event.data.get("agent_key", self._profile.default_remote_agent)
	146	+ return RouteTarget(type="remote_agent", agent_key=agent_key)
	147	+ return RouteTarget(type="requester")
	148	+
	149	+
	150	+class RemoteAgentSourceStrategy(RouteStrategy):
	151	+ """Remote Agent 源路由策略
	152	+
	153	+ NOTE: 当前业务代码中尚未有构建 NormalizedEvent(source=remote_agent) 的路径，
	154	+ VA 事件仍由 RemoteAgentHandler 直接 enqueue_event，未走 dispatch 路由。
	155	+ 本策略为预留设计，待 VA 事件统一归一化后走 dispatch 时激活：
	156	+ - 终态帧（CONTROL_COMPLETED / CONTROL_FAILED）→ 路由到 local_agent
	157	+ - 非终态帧（DATA / CONTROL_WORKING 等）→ 路由到 channel
	158	+ """
	159	+
	160	+ def __init__(self, profile: RemoteAgentSourceProfile):
	161	+ self._profile = profile
	162	+ self._terminal_frame_types: Set[str] = set(profile.terminal_frame_types)
	163	+ self._frame_type_map: Dict[str, str] = profile.frame_type_map
	164	+
	165	+ async def route(self, event: NormalizedEvent, context: RouteContext) -> RouteTarget:
	166	+ frame_type = self._classify_frame(event)
	167	+ if frame_type in self._terminal_frame_types:
	168	+ return RouteTarget(type="local_agent")
	169	+ return RouteTarget(type="requester")
	170	+
	171	+ def _classify_frame(self, event: NormalizedEvent) -> str:
	172	+ if "frame_type" in event.metadata:
	173	+ return event.metadata["frame_type"]
	174	+ return self._frame_type_map.get(event.type.upper(), self._profile.default_frame_type)

Aapplications/a2a_service/orchestrator/state/__init__.py+9-0

@@ -0,0 +1,9 @@
	1	+from .task_state_manager import (
	2	+ CONV_TASK_KEY,
	3	+ InputRequiredState,
	4	+ META_KEY_REMOTE_TASK_ID,
	5	+ META_KEY_SOURCE_AGENT,
	6	+ META_KEY_SUB_TASKS,
	7	+ TaskStateManager,
	8	+)
	9	+from .frame_type import FrameType, classify_frame

Aapplications/a2a_service/orchestrator/state/frame_type.py+42-0

@@ -0,0 +1,42 @@
	1	+from __future__ import annotations
	2	+
	3	+from enum import Enum
	4	+from typing import Optional
	5	+
	6	+from ..route.normalized_event import NormalizedEvent
	7	+
	8	+
	9	+class FrameType(str, Enum):
	10	+ DATA = "DATA"
	11	+ CONTROL_UNSPECIFIED = "CONTROL_UNSPECIFIED"
	12	+ CONTROL_SUBMITTED = "CONTROL_SUBMITTED"
	13	+ CONTROL_WORKING = "CONTROL_WORKING"
	14	+ CONTROL_COMPLETED = "CONTROL_COMPLETED"
	15	+ CONTROL_CANCELED = "CONTROL_CANCELED"
	16	+ CONTROL_FAILED = "CONTROL_FAILED"
	17	+ CONTROL_INPUT_REQUIRED = "CONTROL_INPUT_REQUIRED"
	18	+ CONTROL_REJECTED = "CONTROL_REJECTED"
	19	+ CONTROL_AUTH_REQUIRED = "CONTROL_AUTH_REQUIRED"
	20	+
	21	+
	22	+def classify_frame(event: NormalizedEvent) -> FrameType:
	23	+ if "frame_type" in event.metadata:
	24	+ ft = event.metadata["frame_type"]
	25	+ try:
	26	+ return FrameType(ft)
	27	+ except ValueError:
	28	+ return FrameType.CONTROL_UNSPECIFIED
	29	+
	30	+ event_type = event.type.upper()
	31	+ mapping = {
	32	+ "ARTIFACT": FrameType.DATA,
	33	+ "COMPLETED": FrameType.CONTROL_COMPLETED,
	34	+ "FAILED": FrameType.CONTROL_FAILED,
	35	+ "INPUT_REQUIRED": FrameType.CONTROL_INPUT_REQUIRED,
	36	+ "AUTH_REQUIRED": FrameType.CONTROL_AUTH_REQUIRED,
	37	+ "WORKING": FrameType.CONTROL_WORKING,
	38	+ "SUBMITTED": FrameType.CONTROL_SUBMITTED,
	39	+ "CANCELED": FrameType.CONTROL_CANCELED,
	40	+ "REJECTED": FrameType.CONTROL_REJECTED,
	41	+ }
	42	+ return mapping.get(event_type, FrameType.CONTROL_UNSPECIFIED)