已合并
GH杭研现场问题修改 #349
guoyangsen创建于 6月29日
GH杭研现场问题修改 #349
已合并
共 1 个文件变更+38-7
| @@ -30,6 +30,9 @@ from a2a.server.agent_execution import AgentExecutor, RequestContext | |||
| 30 | from a2a.server.context import ServerCallContext | 30 | from a2a.server.context import ServerCallContext |
| 31 | from a2a.server.events import EventQueue | 31 | from a2a.server.events import EventQueue |
| 32 | from a2a.types.a2a_pb2 import ( | 32 | from a2a.types.a2a_pb2 import ( |
| 33 | + AgentCapabilities, | ||
| 34 | + AgentCard, | ||
| 35 | + AgentInterface, | ||
| 33 | CancelTaskRequest, | 36 | CancelTaskRequest, |
| 34 | GetTaskRequest, | 37 | GetTaskRequest, |
| 35 | Message, | 38 | Message, |
| @@ -49,6 +52,7 @@ from a2a.types.a2a_pb2 import ( | |||
| 49 | TASK_STATE_INPUT_REQUIRED, | 52 | TASK_STATE_INPUT_REQUIRED, |
| 50 | TASK_STATE_WORKING, | 53 | TASK_STATE_WORKING, |
| 51 | ) | 54 | ) |
| 55 | +from a2a.utils.constants import PROTOCOL_VERSION_1_0, TransportProtocol | ||
| 52 | from a2a.utils.errors import TaskNotFoundError, UnsupportedOperationError | 56 | from a2a.utils.errors import TaskNotFoundError, UnsupportedOperationError |
| 53 | from google.protobuf.json_format import MessageToDict, ParseDict | 57 | from google.protobuf.json_format import MessageToDict, ParseDict |
| 54 | from google.protobuf.struct_pb2 import Struct, Value | 58 | from google.protobuf.struct_pb2 import Struct, Value |
| @@ -357,15 +361,38 @@ class Executor(AgentExecutor): | |||
| 357 | """是否具备派发能力：注入了单 client（兜底）或有 client_factory（按 url 构造）。""" | 361 | """是否具备派发能力：注入了单 client（兜底）或有 client_factory（按 url 构造）。""" |
| 358 | return self._sub_agent_client is not None or self._client_factory is not None | 362 | return self._sub_agent_client is not None or self._client_factory is not None |
| 359 | 363 | ||
| 364 | + | ||
| 365 | + def _build_sub_agent_card(url: str) -> AgentCard: | ||
| 366 | + """用主侧已知可达地址 url（yaml 配置的 spec.url）自建子 Agent 最小 Card。 | ||
| 367 | + | ||
| 368 | + 不信任子 Agent Card 自报的 netloc——容器内 | ||
| 369 | + ``host:8090`` 会让 RPC 回环到主侧自身（见部署排障）。改为以 yaml 配置的 | ||
| 370 | + 可达地址为权威 RPC 目标，对齐 VA client（``_build_va_card`` + ``create``）。 | ||
| 371 | + | ||
| 372 | + 子 Agent A2A RPC 端点挂在 ``/a2a/``，补尾斜杠避免 ``POST /a2a`` 触发 307。 | ||
| 373 | + """ | ||
| 374 | + rpc_url = url.rstrip("/") + "/" | ||
| 375 | + card = AgentCard(name="SubDPA", description="子 Agent A2A 端点", version="1.0.0") | ||
| 376 | + card.supported_interfaces.append( | ||
| 377 | + AgentInterface( | ||
| 378 | + protocol_binding=TransportProtocol.JSONRPC, | ||
| 379 | + url=rpc_url, | ||
| 380 | + protocol_version=PROTOCOL_VERSION_1_0, | ||
| 381 | + ) | ||
| 382 | + ) | ||
| 383 | + card.capabilities.CopyFrom(AgentCapabilities(streaming=True)) | ||
| 384 | + return card | ||
| 385 | + | ||
| 360 | async def _get_sub_agent_client(self, url: str) -> Optional[Client]: | 386 | async def _get_sub_agent_client(self, url: str) -> Optional[Client]: |
| 361 | """解析某 url 对应的子 Agent A2A client。 | 387 | """解析某 url 对应的子 Agent A2A client。 |
| 362 | 388 | ||
| 363 | - 注入了 ``sub_agent_client``（单测/单部署兜底）→ 直接复用，忽略 url 差异； | 389 | - 注入了 ``sub_agent_client``（单测/单部署兜底）→ 直接复用，忽略 url 差异； |
| 364 | - - 否则用 ``client_factory`` 按 url 懒 ``create_from_url`` 并缓存（每 url 一个）； | 390 | + - 否则用 ``client_factory`` 按 yaml 配置的 url 自建 Card 后 ``create`` 并缓存（每 url 一个）； |
| 365 | - url 为空或无 factory → 返回 None（由调用方将该实体降级 failed）。 | 391 | - url 为空或无 factory → 返回 None（由调用方将该实体降级 failed）。 |
| 366 | 392 | ||
| 367 | - ``create_from_url`` 可能抛（子 Agent 不可达 / 卡片拉取失败）；不在此吞掉， | 393 | + 直接用 yaml 中的 url 作为 RPC 目标，不再 ``create_from_url`` 拉卡片 |
| 368 | - 由 ``_run_sub_agent`` 的 try/except 捕获 → 仅该实体 failed，不连坐其他。 | 394 | + （子 Agent Card 自报容器内地址会导致 RPC 回环到主侧）。建连/发送失败由 |
| 395 | + ``_run_sub_agent`` 的 try/except 捕获 → 仅该实体 failed，不连坐其他。 | ||
| 369 | """ | 396 | """ |
| 370 | if self._sub_agent_client is not None: | 397 | if self._sub_agent_client is not None: |
| 371 | return self._sub_agent_client | 398 | return self._sub_agent_client |
| @@ -373,9 +400,10 @@ class Executor(AgentExecutor): | |||
| 373 | return None | 400 | return None |
| 374 | client = self._sub_agent_clients.get(url) | 401 | client = self._sub_agent_clients.get(url) |
| 375 | if client is None: | 402 | if client is None: |
| 376 | - client = await self._client_factory.create_from_url(url) | 403 | + card = self._build_sub_agent_card(url) |
| 404 | + client = self._client_factory.create(card) | ||
| 377 | self._sub_agent_clients[url] = client | 405 | self._sub_agent_clients[url] = client |
| 378 | - logger.info(f"[Executor] sub_agent client 懒构造并缓存：url={url}") | 406 | + logger.info(f"[Executor] sub_agent client 构造并缓存（RPC 直连 yaml 地址）：url={url}") |
| 379 | return client | 407 | return client |
| 380 | 408 | ||
| 381 | async def cancel_task(self, conv_id: str) -> None: | 409 | async def cancel_task(self, conv_id: str) -> None: |
| @@ -1224,7 +1252,7 @@ class Executor(AgentExecutor): | |||
| 1224 | "trace_id": session_ctx.get("trace_id", ""), # 链路可观测性 | 1252 | "trace_id": session_ctx.get("trace_id", ""), # 链路可观测性 |
| 1225 | "agent_id": getattr(get_settings(), "dpa_agent_id", "") or "", # 子 Agent 自身 ID | 1253 | "agent_id": getattr(get_settings(), "dpa_agent_id", "") or "", # 子 Agent 自身 ID |
| 1226 | "params": session_ctx.get("params", {}), # 继承父请求 URL query（如 type/workspace_id），供子 Agent 工作流调用 VA 时透传 | 1254 | "params": session_ctx.get("params", {}), # 继承父请求 URL query（如 type/workspace_id），供子 Agent 工作流调用 VA 时透传 |
| 1227 | - "body": {}, # 子 Agent 自行构造 VA 请求体，初始为空 | 1255 | + "body": session_ctx.get("body", {}), # 继承父请求体：子 Agent 调 VA 时仅覆盖 query/intent，其余业务字段沿用父请求 |
| 1228 | }, | 1256 | }, |
| 1229 | ex=_TTL, | 1257 | ex=_TTL, |
| 1230 | ) | 1258 | ) |
| @@ -1362,7 +1390,9 @@ class Executor(AgentExecutor): | |||
| 1362 | ) -> SubAgentResult: | 1390 | ) -> SubAgentResult: |
| 1363 | """驱动单个子 Agent：node_start → 盖章/透传 report → node_end，并捕获 child_task_id。""" | 1391 | """驱动单个子 Agent：node_start → 盖章/透传 report → node_end，并捕获 child_task_id。""" |
| 1364 | conv_id = turn_ctx.conv_id | 1392 | conv_id = turn_ctx.conv_id |
| 1365 | - sub_conv_id = f"{conv_id}:sub:{spec.entity_id}" | 1393 | + # 分隔符用 '-' 而非 ':'：子 Agent 会把该 conv_id 透传给 VA 作 conversationId， |
| 1394 | + # 工作流平台校验 conversationId 必须匹配 ^[a-zA-Z0-9_-]+$（冒号会被拒，返回 500）。 | ||
| 1395 | + sub_conv_id = f"{conv_id}-sub-{spec.entity_id}" | ||
| 1366 | await self._emit_sub_task( | 1396 | await self._emit_sub_task( |
| 1367 | turn_ctx, child_path, "agent", | 1397 | turn_ctx, child_path, "agent", |
| 1368 | {"event": "node_start", "entity_name": spec.entity_name}, | 1398 | {"event": "node_start", "entity_name": spec.entity_name}, |
| @@ -1518,6 +1548,7 @@ class Executor(AgentExecutor): | |||
| 1518 | "headers": parent_cached.get("headers", {}), # VA 鉴权 token | 1548 | "headers": parent_cached.get("headers", {}), # VA 鉴权 token |
| 1519 | "params": parent_cached.get("params", {}), # 前端 URL query（如 type/workspace_id）继承给子 Agent 的 VA 调用 | 1549 | "params": parent_cached.get("params", {}), # 前端 URL query（如 type/workspace_id）继承给子 Agent 的 VA 调用 |
| 1520 | "trace_id": parent_cached.get("trace_id", ""), # 链路追踪 | 1550 | "trace_id": parent_cached.get("trace_id", ""), # 链路追踪 |
| 1551 | + "body": parent_cached.get("body", {}), # 下传父请求体：子 Agent 仅覆盖 query/intent，其余业务字段继承（工作流 code 组件依赖） | ||
| 1521 | "sub_task_path": list(child_path), # 下传绝对路径：子节点继续盖章 / 深度门控 | 1552 | "sub_task_path": list(child_path), # 下传绝对路径：子节点继续盖章 / 深度门控 |
| 1522 | } | 1553 | } |
| 1523 | request = self._build_sub_agent_message(spec.query, sub_conv_id, session_context) | 1554 | request = self._build_sub_agent_message(spec.query, sub_conv_id, session_context) |