From 6a8c4fd95856ec5da3876b9eecbb1d829f0a5d14 Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Mon, 17 Aug 2026 17:06:38 +0800 Subject: [PATCH] docs: fix 18 doc-code gaps, convert all diagrams to Mermaid MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix documentation-vs-code gaps identified in deep analysis: HIGH: - README: move LlmHelper to samples module (was listed under client) - API_REFERENCE (en+zh): add missing core package section - API_REFERENCE + DEVELOPER_GUIDE (en+zh): add TASK_STATUS_CHANGED and WORKFLOW_COMPLETE EventType constants (code has 22, docs had 20) - CHANGELOG: replace stale ExtensionCallback reference with ExtensionSender - pom.xml: fix SCM URL from personal fork to project-openan upstream MEDIUM: - DESIGN (en+zh): remove duplicate Section 4, fix numbering 7->8->9 - DEVELOPER_GUIDE (en+zh): fix checklist (3 required -> 2 required + 2 defaults) - DEVELOPER_GUIDE (en+zh): fix broken code formatting in AuthProvider section - API_REFERENCE (en+zh): add ConversationScopedA2AJavaClientRuntime - API_REFERENCE (en+zh): add TaskStatus enum (PENDING/RUNNING/SUCCESS/FAILED) - API_REFERENCE (en+zh): add AgentCardJacksonModule and AgentCardNormalizer - API_REFERENCE (en+zh): add spring-boot-starter module documentation - notification-t-design-analysis (en+zh): fix 5 broken Mermaid blocks, fix stale file paths LOW: - README: add missing Chinese doc links - DESIGN: replace asyncio.gather (Python) with CompletableFuture (Java) - README + DESIGN: align architecture layer descriptions (add Foundation layer) Diagrams: - Convert all ASCII art diagrams to Mermaid format across: README.md, docs/zh/DESIGN.md, docs/zh/工作台AgentScope结合方案.md Signed-off-by: Zhoujie628 --- CHANGELOG.md | 3 +- README.md | 65 +++++--- docs/en/API_REFERENCE.md | 150 ++++++++++++++++++ docs/en/DESIGN.md | 12 +- docs/en/DEVELOPER_GUIDE.md | 21 +-- docs/en/notification-t-design-analysis.md | 13 +- docs/zh/API_REFERENCE.md | 150 ++++++++++++++++++ docs/zh/DESIGN.md | 22 +-- docs/zh/DEVELOPER_GUIDE.md | 14 +- docs/zh/notification-t-design-analysis.md | 4 - ...23\345\220\210\346\226\271\346\241\210.md" | 35 ++-- pom.xml | 6 +- 12 files changed, 393 insertions(+), 102 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 215f7a96..a78cf6de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,8 +18,7 @@ and full A2A-T extension support. Full A2A-T extension support and multi-protoco extraction - `ExtensionSender` / `DefaultExtensionSender`: one-shot pre-positioning facade for Authorization-T and Notification-T (long-lived SSE subscription) -- `ControlPoint` / `ExtensionCallback` split: flow decisions (`onTask` / `onSelfTask` / `onRoute` / `onNegotiation`) and - reactive hooks (`onAuthorization` / `onNotification`) on separate interfaces +- `ControlPoint` / `ExtensionSender` split: flow decisions (`onTask` / `onSelfTask` / `onRoute` / `onNegotiation`) on `ControlPoint`; Authorization-T / Notification-T pre-positioning on `ExtensionSender` - `NegotiationStrategy`: pluggable clarification strategy injected into `DefaultControlPoint` - `SELF_LOOP` step type for local task handling without an A2A-T message diff --git a/README.md b/README.md index db1e420a..a6e34468 100644 --- a/README.md +++ b/README.md @@ -92,13 +92,15 @@ System.out.println("Success: " + result.isSuccess()); ## Architecture -``` -Layer 2 - Orchestration ExecutePsop - | lifecycle, event stream, onFinish hook -Layer 1 - Traversal WorkflowExecutor - | DAG walk, parallel dispatch, context assembly, routing -Layer 0 - Transport WorkflowEngineClient / A2ATransport - A2A send, auth, extensions, SSL, SSE +```mermaid +graph TD + L2["Layer 2 — Orchestration
ExecutePsop
lifecycle, event stream, onFinish hook"] + L1["Layer 1 — Traversal
WorkflowExecutor
DAG walk, parallel dispatch, context assembly, routing"] + L0["Layer 0 — Transport
WorkflowEngineClient / A2ATransport
A2A send, auth, extensions, SSL, SSE"] + F["Foundation — Decision
ControlPoint
user-implemented business decisions"] + + L2 --> L1 --> L0 + L0 -.-> F ``` | Layer | Entry Point | Responsibility | @@ -106,29 +108,38 @@ Layer 0 - Transport WorkflowEngineClient / A2ATransport | High | `ExecutePsop.Builder` | Event stream, lifecycle, `onFinish` persistence | | Mid | `WorkflowExecutor` | DAG traversal, context assembly, ControlPoint dispatch | | Low | `WorkflowEngineClient` / `A2ATransport` | A2A send, auth, extensions, SSL, SSE normalization | +| Foundation | `ControlPoint` | User-implemented business decisions (onTask, onSelfTask, onRoute, onNegotiation) | ## Package Structure +```mermaid +graph TD + root["dev.openan.workflow.engine"] + client["client
A2A transport, auth, extensions"] + control["control
User decision interfaces"] + core["core
DAG traversal, context assembly"] + model["model
Data models"] + registry["registry
LoadPsop, RegistryClient"] + runner["runner
ExecutePsop entry point"] + + root --> client + root --> control + root --> core + root --> model + root --> registry + root --> runner ``` -dev.openan.workflow.engine -├── client # A2A transport, auth, extensions (package-private internals) -│ ├── WorkflowEngineClient # Send facade interface -│ ├── DefaultWorkflowEngineClient # Send + Negotiation-T auto-loop -│ ├── ExtensionSender # One-shot pre-positioning (Auth-T, Notification-T) -│ ├── A2ATransport # Shared wire layer (runtime, auth, SSE) -│ ├── AuthProvider # Custom auth provider interface -│ ├── WorkflowEngineClientConfig # Builder config: SSL, auth, A2A-T -│ ├── LlmHelper # LLM utility (wraps a2a-t-sdk LLMClient) -│ └── CredentialCrypto # AES-256-GCM credential encryption -├── control # User-facing decision interfaces -│ ├── ControlPoint # onTask, onSelfTask, onRoute, onNegotiation -│ ├── DefaultControlPoint # Default routing implementation -│ ├── EventCallback # Event stream callback -│ └── NegotiationStrategy # Pluggable negotiation clarification -├── model # Data models (Workflow, Task, StepType, etc.) -├── registry # LoadPsop, RegistryClient -└── runner # ExecutePsop (entry point) -``` + +| Package | Key Classes | Description | +|---------|-------------|-------------| +| `client` | `WorkflowEngineClient`, `DefaultWorkflowEngineClient`, `ExtensionSender`, `A2ATransport`, `AuthProvider`, `AgentAuthManager`, `WorkflowEngineClientConfig`, `CredentialCrypto`, `AgentCardJacksonModule` | A2A transport, auth, extensions (package-private internals) | +| `control` | `ControlPoint`, `DefaultControlPoint`, `EventCallback`, `EventType`, `NegotiationStrategy` | User-facing decision interfaces | +| `core` | `WorkflowExecutor`, `ContextBuilder` | DAG traversal, context assembly | +| `model` | `Workflow`, `WorkflowStep`, `Task`, `TaskRequest`, `TaskResponse`, `StepType`, `JumpCondition`, `RouteDecision`, `ExecutionResult`, `SendMessageResult`, `TaskStatus`, `WorkflowSearchResult` | Data models | +| `registry` | `LoadPsop`, `RegistryClient` | PSOP loading and AgentCard registry | +| `runner` | `ExecutePsop` | Entry point for workflow execution | + +> **Note:** `LlmHelper` lives in the **samples** module (`dev.openan.workflow.engine.examples`), not in the `client` package. The workflow engine itself does not call an LLM directly. ## Documentation @@ -149,6 +160,8 @@ dev.openan.workflow.engine - [Notification-T 设计分析](docs/zh/notification-t-design-analysis.md) - 长连接 SSE 设计选型 - [业务流](docs/zh/业务流.md) - SPN 跨城诊断业务流程 - [调用过程](docs/zh/调用过程.md) - 端到端报文交互示例 +- [工作台集成指南](docs/zh/工作台集成指南.md) - 工作台 Spring Boot 集成完整指南 +- [工作台 AgentScope 结合方案](docs/zh/工作台AgentScope结合方案.md) - AgentScope 框架结合方案 ## Modules diff --git a/docs/en/API_REFERENCE.md b/docs/en/API_REFERENCE.md index f194cf64..f007b9dd 100644 --- a/docs/en/API_REFERENCE.md +++ b/docs/en/API_REFERENCE.md @@ -6,6 +6,7 @@ |-----------------------------------|-------------------------------------------------| | `dev.openan.workflow.engine.client` | A2A message transport, auth, extensions, config | | `dev.openan.workflow.engine.control` | User decision points and event system | +| `dev.openan.workflow.engine.core` | DAG traversal engine and context assembly | | `dev.openan.workflow.engine.model` | Data models (Workflow, Task, results) | | `dev.openan.workflow.engine.registry` | PSOP loading and AgentCard registry | | `dev.openan.workflow.engine.runner` | Entry point for workflow execution | @@ -234,6 +235,55 @@ public interface A2AJavaClientRuntime { A default implementation is provided. Implement this interface only if you need custom HTTP transport. +### ConversationScopedA2AJavaClientRuntime + +Optional lifecycle callback for runtime implementations whose transport session spans multiple A2A requests (e.g. a gateway login that must stay alive across all negotiation rounds). + +```java +public interface ConversationScopedA2AJavaClientRuntime { + void closeConversation(AgentCard agentCard, String contextId); +} +``` + +When the runtime also implements this interface, the engine calls `closeConversation` after the full send + negotiation cycle completes — not after each individual HTTP request. This allows gateway sessions to be released only after the logical conversation is done. + +Implement alongside `A2AJavaClientRuntime`: + +```java +public class MyGatewayRuntime + implements A2AJavaClientRuntime, ConversationScopedA2AJavaClientRuntime { + @Override + public Iterable sendMessage(...) { ... } + + @Override + public void closeConversation(AgentCard agentCard, String contextId) { + // release gateway session + } + + @Override + public void close() { ... } +} +``` + +### AgentCardJacksonModule + +Jackson module for deserializing AgentCard JSON with security scheme normalization. Handles the OpenAPI-format `securitySchemes` / `securityRequirements` fields that the A2A SDK's strongly-typed `AgentCard` record expects. + +```java +ObjectMapper mapper = new ObjectMapper() + .registerModule(new AgentCardJacksonModule()); +AgentCard card = mapper.readValue( + new File("agentcard/my_agent.json"), AgentCard.class); +``` + +### AgentCardNormalizer + +Utility that normalizes a raw `Map` (as returned by the Registry Center API) into the `AgentCard`-compatible format. Used internally by `RegistryClient.fetchAgentCards()`. Also available as a public static method for custom normalization: + +```java +Map normalized = AgentCardNormalizer.normalize(rawMap); +``` + --- ## dev.openan.workflow.engine.control @@ -300,6 +350,7 @@ Override to receive real-time execution events. Event types are defined in `Even | `STEP_COMPLETE` | A workflow step completed | | `TASK_REQUEST` | A task was dispatched to an agent | | `TASK_RESPONSE` | A task response was received | +| `TASK_STATUS_CHANGED` | A task's status changed (pending → running → success/failed) | | `AGENT_REQUEST` | A message was sent to an agent | | `AGENT_RESPONSE` | A response was received from an agent | | `AGENT_STATUS_UPDATE` | Agent SSE status update (SUBMITTED, WORKING, etc.) | @@ -312,6 +363,7 @@ Override to receive real-time execution events. Event types are defined in `Even | `AUTHORIZATION_RESOLVED` | Authorization decision was made | | `NOTIFICATION` | Notification received from agent | | `ROUTE_DECISION` | Route decision was made | +| `WORKFLOW_COMPLETE` | The workflow completed (all steps finished) | | `START` | Workflow execution started | | `COMPLETE` | Workflow execution completed successfully | | `ERROR` | Workflow execution failed | @@ -371,6 +423,27 @@ Map registerAgentCard(Map agentCard) --- +## dev.openan.workflow.engine.core + +### WorkflowExecutor + +Mid-layer DAG traversal engine. Walks the workflow steps, assembles upstream context via `ContextBuilder`, dispatches subtasks in parallel, applies step success policies (`ALL_SUCCESS` / `ANY_SUCCESS` / `SELF_LOOP`), and routes to the next step. + +Not typically instantiated directly by SDK users — `ExecutePsop` wraps it internally. Exposed for advanced integrations that need to run the traversal layer without the runner's lifecycle management. + +```java +WorkflowExecutor executor = new WorkflowExecutor( + workflow, controlPoint, engineClient, + eventCallback, runtimeIntent, lang); +ExecutionResult result = executor.run().join(); +``` + +### ContextBuilder + +Package-private helper that assembles the context message for each step by folding upstream step outputs according to `contextFrom` rules (`"*"` = all ancestors, specific names = selective inheritance). Not part of the public API surface. + +--- + ## dev.openan.workflow.engine.model ### Workflow @@ -403,6 +476,17 @@ Static factory: `Workflow.fromMap(Map)` parses from orchestratio | `ANY_SUCCESS` | Any subtask success is sufficient | | `SELF_LOOP` | The workflow agent handles the task locally via `onSelfTask`; no A2A-T message is sent. Success follows `ALL_SUCCESS`. | +### TaskStatus + +Task lifecycle status, used in `TASK_STATUS_CHANGED` events for cross-SDK consistency with the Python SDK. + +| Value | String | Description | +|-----------|------------|--------------------------------------| +| `PENDING` | `"pending"` | Task created, not yet started | +| `RUNNING` | `"running"` | Task in progress | +| `SUCCESS` | `"success"` | Task completed successfully | +| `FAILED` | `"failed"` | Task failed | + ### Task | Field | Type | Description | @@ -511,3 +595,69 @@ Static factory: `Workflow.fromMap(Map)` parses from orchestratio - Negotiation failures fall through after `maxNegotiationRounds`. - Auth failures (401) are logged as `ERROR` and auth headers are not set; the request proceeds without auth. - SSE stream errors after terminal events are logged at `DEBUG` level (expected behavior). + +--- + +## spring-boot-starter Module + +The `spring-boot-starter` module provides Spring Boot auto-configuration for the A2A **server** side (not the client/workflow side). When on the classpath of a Spring Boot web application, it auto-registers all A2A SDK server components as Spring beans. + +### A2AProperties + +Configuration properties prefixed with `a2at.server`: + +| Property | Default | Description | +|----------|---------|-------------| +| `a2at.server.agent-card` | `classpath:agentcard.json` | Path to the AgentCard JSON file (classpath: or file: prefix supported) | +| `a2at.server.path-prefix` | `/a2a/json` | URL path prefix for A2A endpoints | + +```yaml +a2at: + server: + agent-card: classpath:agentcard/my_agent.json + path-prefix: /a2a/json +``` + +### A2AAutoConfiguration + +Auto-configures the following beans (all `@ConditionalOnMissingBean`, so you can override any): + +| Bean | Type | Purpose | +|------|------|---------| +| `agentCard` | `AgentCard` | Loaded from `a2at.server.agent-card` path via Jackson | +| `a2aConfigProvider` | `A2AConfigProvider` | SDK configuration values | +| `taskStore` | `InMemoryTaskStore` | In-memory task storage | +| `eventBus` | `MainEventBus` | Event bus for SSE streaming | +| `queueManager` | `InMemoryQueueManager` | Event queue manager | +| `pushStore` | `PushNotificationConfigStore` | Push notification config storage | +| `agentExecutorPool` | `ExecutorService` | Thread pool for agent execution (8 threads, daemon) | +| `eventBusProcessor` | `MainEventBusProcessor` | Event bus processor | +| `requestHandler` | `RequestHandler` | Default request handler | +| `restHandler` | `RestHandler` | REST protocol handler | +| `a2aController` | `A2AController` | Spring MVC controller (`message:send` + `message:stream`) | + +### A2AController + +Spring MVC controller that exposes A2A REST endpoints: + +- `POST {path-prefix}/message:send` — blocking send +- `POST {path-prefix}/message:stream` — SSE streaming + +### Usage + +The partner only needs to provide an `AgentExecutor` implementation: + +```java +@Component +public class MyAgentExecutor implements AgentExecutor { + @Override + public ExecuteResult execute(ExecuteRequest request) { + // business logic + return ExecuteResult.builder() + .addTextPart("result text") + .build(); + } +} +``` + +All other beans (AgentCard, RequestHandler, RestHandler, A2AController, etc.) are auto-configured. diff --git a/docs/en/DESIGN.md b/docs/en/DESIGN.md index cba9ac1d..9f606f2c 100644 --- a/docs/en/DESIGN.md +++ b/docs/en/DESIGN.md @@ -80,8 +80,7 @@ concurrently, applies the step's success policy, and determines the next step (s Step dispatch rules: -- Steps whose predecessors are all satisfied are collected and dispatched in parallel (`asyncio.gather` / - `CompletableFuture`), so steps at the same layer run concurrently. +- Steps whose predecessors are all satisfied are collected and dispatched in parallel (`CompletableFuture`), so steps at the same layer run concurrently. - Subtasks within a step also run in parallel. - `ALL_SUCCESS` - all subtasks must succeed. - `ANY_SUCCESS` - the first successful subtask wins; the rest are cancelled. @@ -115,11 +114,6 @@ Authorization-T and Notification-T are pre-positioning concerns handled via `Ext --- -## 4. A2A-T Extension Model -means a host that only cares about routing does not have to implement (or stub) authorization hooks, and vice versa. - ---- - ## 4. A2A-T Extension Model Four A2A-T extensions are supported. They divide into two groups by lifecycle. @@ -269,7 +263,7 @@ sequenceDiagram --- -## 9. Dependencies +## 8. Dependencies **Java SDK:** `org.a2aproject.sdk:a2a-java-sdk-client` (A2A protocol), @@ -279,7 +273,7 @@ The SDK is standalone: it does not depend on the orchestration center. --- -## 10. Design Decisions Summary +## 9. Design Decisions Summary 1. **Shared transport, two facades** - wire machinery written once on `A2ATransport`; `WorkflowEngineClient` and `ExtensionSender` each own one orchestration concern and delegate wire diff --git a/docs/en/DEVELOPER_GUIDE.md b/docs/en/DEVELOPER_GUIDE.md index 1d79ab85..e4a06371 100644 --- a/docs/en/DEVELOPER_GUIDE.md +++ b/docs/en/DEVELOPER_GUIDE.md @@ -94,8 +94,10 @@ Events come from three layers: the runner (lifecycle bracket), the executor (ste | `step_start` | executor | Step begins | `step` | | `task_request` | executor | A subtask is dispatched to `onTask`/`onSelfTask` | `step`, `agent`, `task` | | `task_response` | executor | `onTask`/`onSelfTask` returned a `TaskResponse` | `step`, `agent`, `task`, `output` | +| `task_status_changed` | executor | Task status changed (pending → running → success/failed) | `step`, `agent`, `task`, `status` | | `route_decision` | executor | Branch chosen | `step`, `next`, `reason` | | `step_complete` | executor | Step finished | `step`, `results` | +| `workflow_complete` | executor | All steps finished | `history`, `step_outputs` | | `agent_request` | engine client | Message sent to agent | `agent`, `request`, `metadata` | | `agent_response` | engine client | Response from agent | `agent`, `response` | | `agent_status_update` | engine client | Agent SSE status update | `agent`, `state`, `is_final` | @@ -205,22 +207,13 @@ Passwords can be AES-GCM encrypted with `enc::` prefix. The decr ### 7.2 Custom AuthProvider For non-standard auth (SSO, API keys, custom headers): - ```java WorkflowEngineClientConfig.builder() - . - -authProvider((agentName, agentCard, headers) ->{ - headers. - -put("Authorization","Bearer "+mySsoToken); - headers. - -put("X-Custom","value"); + .authProvider((agentName, agentCard, headers) -> { + headers.put("Authorization", "Bearer " + mySsoToken); + headers.put("X-Custom", "value"); }) - . - -build(); + .build(); ``` ### 7.3 Credential File Fields @@ -300,7 +293,7 @@ subscriber and let the future complete. ## 11. Checklist 1. Add Maven dependencies -2. Implement `ControlPoint` (onTask + onSelfTask + onRoute) +2. Implement `ControlPoint` (at minimum `onTask` + `onRoute`; `onSelfTask` and `onNegotiation` have defaults) 3. Get AgentCards (from registry or JSON files) 4. Load Workflow (via `LoadPsop` or build your own) 5. Configure `.env` and credentials file diff --git a/docs/en/notification-t-design-analysis.md b/docs/en/notification-t-design-analysis.md index 0a2521ca..dad4d835 100644 --- a/docs/en/notification-t-design-analysis.md +++ b/docs/en/notification-t-design-analysis.md @@ -18,6 +18,7 @@ The engine currently sends Authorization-T, Notification-T, and Task-T as three calls. Each opens its own HTTP connection (SSE stream), gets a response, and closes. ``` + ```mermaid sequenceDiagram participant W as Workbench @@ -37,7 +38,6 @@ sequenceDiagram S-->>W: artifact (recovery result) S-->>W: COMPLETED (stream closes) ``` -``` ### Key Code Locations @@ -105,6 +105,7 @@ stream open. When recovery completes, the SPN Agent pushes the result through th disconnecting. ``` + ```mermaid sequenceDiagram participant W as Workbench @@ -119,7 +120,6 @@ sequenceDiagram S-->>W: SSE event: recovery result (via Notification-T stream) W->>S: close connection (cancel subscription) ``` -``` ### Authentication @@ -192,15 +192,14 @@ This pattern is correct and requires no modification. - SPN agent city2: samples/src/main/java/dev/openan/workflow/engine/examples/agents/SpnDomainAgentCity2Executor.java - Workbench control: samples/src/main/java/dev/openan/workflow/engine/examples/agents/WorkbenchControlPoint.java - Demo entry: samples/src/main/java/dev/openan/workflow/engine/examples/SpnCrossCityDiagnosisDemo.java -- Agent server: samples/src/main/java/dev/openan/workflow/engine/examples/StartAgentsServer.java +- Agent server: samples/src/main/java/dev/openan/workflow/engine/examples/embedded/StartAgentsServer.java - E2E test: samples/src/test/java/dev/openan/workflow/engine/examples/SpnCrossCityE2ETest.java - AgentCard city1: samples/src/main/resources/agentcard/spn_domain_agent_city1.json - AgentCard city2: samples/src/main/resources/agentcard/spn_domain_agent_city2.json - AgentCard workbench: samples/src/main/resources/agentcard/transport_workbench_agent.json - Credentials: samples/src/main/resources/spn_agent_credentials.json -- Workflow JSON: orchestration-center/data/workflow_storage/psop/psop_spn_cross_city_diagnosis.json -- Protocol examples: workflow-exec-engine-java/protocol data example.txt -- Protocol data example: workflow-exec-engine-java/调用过程.md +- Workflow JSON: orchestration-center/data/workflow_storage/psop/psop_spn_cross_city_diagnosis.json (separate repo) +- Protocol examples: docs/zh/调用过程.md ## Business Flow Definition (from business-flow.md / yewuliu.md) @@ -223,6 +222,7 @@ This pattern is correct and requires no modification. ### Complete Business Flow ``` + ```mermaid sequenceDiagram participant U as Upper layer @@ -255,7 +255,6 @@ sequenceDiagram Note over W: SelfLoop: merge W-->>U: merged result ``` -``` ### Key Design Decisions (confirmed) diff --git a/docs/zh/API_REFERENCE.md b/docs/zh/API_REFERENCE.md index 2da03eca..a724d5a7 100644 --- a/docs/zh/API_REFERENCE.md +++ b/docs/zh/API_REFERENCE.md @@ -6,6 +6,7 @@ |-----------------------------------|---------------------------------------| | `dev.openan.workflow.engine.client` | A2A 消息传输、认证、扩展、配置 | | `dev.openan.workflow.engine.control` | 用户决策点和事件系统 | +| `dev.openan.workflow.engine.core` | DAG 遍历引擎和上下文组装 | | `dev.openan.workflow.engine.model` | 数据模型（Workflow、Task、Result 等） | | `dev.openan.workflow.engine.registry` | PSOP 加载和 AgentCard 注册 | | `dev.openan.workflow.engine.runner` | 工作流执行入口 | @@ -230,6 +231,55 @@ public interface A2AJavaClientRuntime { 引擎提供默认实现。仅在需要自定义 HTTP 传输时实现此接口。 +### ConversationScopedA2AJavaClientRuntime + +可选的生命周期回调接口，适用于传输会话跨越多个 A2A 请求的运行时实现（例如网关登录需在所有协商轮次期间保持存活）。 + +```java +public interface ConversationScopedA2AJavaClientRuntime { + void closeConversation(AgentCard agentCard, String contextId); +} +``` + +当运行时同时实现此接口时，引擎会在完整的发送 + 协商周期完成后调用 `closeConversation`——而不是每次 HTTP 请求之后。这样网关会话只会在逻辑会话结束后才释放。 + +与 `A2AJavaClientRuntime` 一起实现： + +```java +public class MyGatewayRuntime + implements A2AJavaClientRuntime, ConversationScopedA2AJavaClientRuntime { + @Override + public Iterable sendMessage(...) { ... } + + @Override + public void closeConversation(AgentCard agentCard, String contextId) { + // 释放网关会话 + } + + @Override + public void close() { ... } +} +``` + +### AgentCardJacksonModule + +用于反序列化 AgentCard JSON 的 Jackson 模块，包含安全方案归一化。处理 A2A SDK 强类型 `AgentCard` record 所需的 OpenAPI 格式 `securitySchemes` / `securityRequirements` 字段。 + +```java +ObjectMapper mapper = new ObjectMapper() + .registerModule(new AgentCardJacksonModule()); +AgentCard card = mapper.readValue( + new File("agentcard/my_agent.json"), AgentCard.class); +``` + +### AgentCardNormalizer + +将原始 `Map`（注册中心 API 返回）归一化为 `AgentCard` 兼容格式的工具类。`RegistryClient.fetchAgentCards()` 内部使用。也作为公共静态方法供自定义归一化使用： + +```java +Map normalized = AgentCardNormalizer.normalize(rawMap); +``` + --- ## dev.openan.workflow.engine.control @@ -296,6 +346,7 @@ public class EventCallback { | `STEP_COMPLETE` | 工作流步骤完成 | | `TASK_REQUEST` | 任务分派给智能体 | | `TASK_RESPONSE` | 收到任务响应 | +| `TASK_STATUS_CHANGED` | 任务状态变更（pending → running → success/failed） | | `AGENT_REQUEST` | 消息发送给智能体 | | `AGENT_RESPONSE` | 收到智能体响应 | | `AGENT_STATUS_UPDATE` | 智能体 SSE 状态更新 | @@ -308,6 +359,7 @@ public class EventCallback { | `AUTHORIZATION_RESOLVED` | 授权决策已做出 | | `NOTIFICATION` | 收到智能体通知 | | `ROUTE_DECISION` | 路由决策已做出 | +| `WORKFLOW_COMPLETE` | 工作流完成（所有步骤结束） | | `START` | 工作流执行开始 | | `COMPLETE` | 工作流执行成功完成 | | `ERROR` | 工作流执行失败 | @@ -367,6 +419,27 @@ Map registerAgentCard(Map agentCard) --- +## dev.openan.workflow.engine.core + +### WorkflowExecutor + +中层 DAG 遍历引擎。遍历工作流步骤，通过 `ContextBuilder` 组装上游上下文，并行分派子任务，应用步骤成功策略（`ALL_SUCCESS` / `ANY_SUCCESS` / `SELF_LOOP`），并路由到下一步。 + +SDK 用户通常不直接实例化——`ExecutePsop` 内部封装了它。供需要在不含 runner 生命周期管理的情况下运行遍历层的高级集成使用。 + +```java +WorkflowExecutor executor = new WorkflowExecutor( + workflow, controlPoint, engineClient, + eventCallback, runtimeIntent, lang); +ExecutionResult result = executor.run().join(); +``` + +### ContextBuilder + +包级私有辅助类，按 `contextFrom` 规则折叠上游步骤输出，为每个步骤组装上下文消息（`"*"` = 所有前驱步骤，指定名称 = 选择性继承）。不属于公共 API 接口。 + +--- + ## dev.openan.workflow.engine.model ### Workflow @@ -399,6 +472,17 @@ Map registerAgentCard(Map agentCard) | `ANY_SUCCESS` | 任一子任务成功即可 | | `SELF_LOOP` | 工作流执行智能体通过 `onSelfTask` 本地处理，不发 A2A-T 消息。成功语义同 `ALL_SUCCESS`。 | +### TaskStatus + +任务生命周期状态，用于 `TASK_STATUS_CHANGED` 事件，与 Python SDK 保持跨 SDK 一致性。 + +| 值 | 字符串 | 说明 | +|------------|------------|------------------------| +| `PENDING` | `"pending"` | 任务已创建，尚未开始 | +| `RUNNING` | `"running"` | 任务进行中 | +| `SUCCESS` | `"success"` | 任务成功完成 | +| `FAILED` | `"failed"` | 任务失败 | + ### Task | 字段 | 类型 | 说明 | @@ -506,3 +590,69 @@ Map registerAgentCard(Map agentCard) - 协商失败在 `maxNegotiationRounds` 轮后终止。 - 认证失败（401）记为 `ERROR` 日志，认证头不设置，请求继续发出。 - SSE 流在终态事件后的连接关闭日志为 `DEBUG` 级别（预期行为）。 + +--- + +## spring-boot-starter 模块 + +`spring-boot-starter` 模块为 A2A **服务端**（非客户端/工作流侧）提供 Spring Boot 自动配置。当位于 Spring Boot Web 应用的 classpath 时，自动将所有 A2A SDK 服务端组件注册为 Spring Bean。 + +### A2AProperties + +以 `a2at.server` 为前缀的配置属性： + +| 属性 | 默认值 | 说明 | +|-------------------------|-------------------------------|-----------------------------------------------| +| `a2at.server.agent-card` | `classpath:agentcard.json` | AgentCard JSON 文件路径（支持 classpath: 或 file: 前缀） | +| `a2at.server.path-prefix` | `/a2a/json` | A2A 端点的 URL 路径前缀 | + +```yaml +a2at: + server: + agent-card: classpath:agentcard/my_agent.json + path-prefix: /a2a/json +``` + +### A2AAutoConfiguration + +自动配置以下 Bean（均为 `@ConditionalOnMissingBean`，可覆盖任意一个）： + +| Bean | 类型 | 用途 | +|---------------------|-------------------------------|---------------------------------------------------| +| `agentCard` | `AgentCard` | 通过 Jackson 从 `a2at.server.agent-card` 路径加载 | +| `a2aConfigProvider` | `A2AConfigProvider` | SDK 配置值 | +| `taskStore` | `InMemoryTaskStore` | 内存任务存储 | +| `eventBus` | `MainEventBus` | 用于 SSE 流的事件总线 | +| `queueManager` | `InMemoryQueueManager` | 事件队列管理器 | +| `pushStore` | `PushNotificationConfigStore` | 推送通知配置存储 | +| `agentExecutorPool` | `ExecutorService` | 智能体执行线程池（8 线程，守护线程） | +| `eventBusProcessor` | `MainEventBusProcessor` | 事件总线处理器 | +| `requestHandler` | `RequestHandler` | 默认请求处理器 | +| `restHandler` | `RestHandler` | REST 协议处理器 | +| `a2aController` | `A2AController` | Spring MVC 控制器（`message:send` + `message:stream`） | + +### A2AController + +暴露 A2A REST 端点的 Spring MVC 控制器： + +- `POST {path-prefix}/message:send` — 阻塞式发送 +- `POST {path-prefix}/message:stream` — SSE 流式 + +### 用法 + +合作方只需提供 `AgentExecutor` 实现： + +```java +@Component +public class MyAgentExecutor implements AgentExecutor { + @Override + public ExecuteResult execute(ExecuteRequest request) { + // 业务逻辑 + return ExecuteResult.builder() + .addTextPart("result text") + .build(); + } +} +``` + +其余 Bean（AgentCard、RequestHandler、RestHandler、A2AController 等）均自动配置。 diff --git a/docs/zh/DESIGN.md b/docs/zh/DESIGN.md index 017c9be1..052bd231 100644 --- a/docs/zh/DESIGN.md +++ b/docs/zh/DESIGN.md @@ -29,15 +29,15 @@ SDK 负责协议机制（消息发送、流式传输、认证、Task-T 提示词 SDK 分为四层，每层构建在下一层之上，单一职责，入口清晰。 -``` -Layer 2 - 编排层 ExecutePsop - | 生命周期、事件流、取消、onFinish 持久化 -Layer 1 - 遍历层 WorkflowExecutor - | DAG 遍历、并行下发、上下文组装、路由 -Layer 0 - 通信层 A2ATransport + 两个门面 - | WorkflowEngineClient（工作流发送）| ExtensionSender（一次性前置下发） -基础层 - 决策 ControlPoint - 用户实现的业务决策 +```mermaid +graph TD + L2["Layer 2 - 编排层
ExecutePsop
生命周期、事件流、取消、onFinish 持久化"] + L1["Layer 1 - 遍历层
WorkflowExecutor
DAG 遍历、并行下发、上下文组装、路由"] + L0["Layer 0 - 通信层
A2ATransport + 两个门面
WorkflowEngineClient（工作流发送）| ExtensionSender（一次性前置下发）"] + F["基础层 - 决策
ControlPoint
用户实现的业务决策"] + + L2 --> L1 --> L0 + L0 -.-> F ``` ### 2.1 Layer 0 - 通信层 @@ -237,7 +237,7 @@ sequenceDiagram --- -## 9. 依赖 +## 8. 依赖 **本 SDK：** `org.a2aproject.sdk:a2a-java-sdk-client`（A2A 协议）、 `net.openan.a2at.sdk:a2a-t-client`（A2A-T 扩展）、Jackson、SLF4J、Lombok。 @@ -246,7 +246,7 @@ SDK 是独立的：不依赖编排中心。 --- -## 10. 设计决策总结 +## 9. 设计决策总结 1. **共享 transport，两个门面** — 通信层机制在 `A2ATransport` 上写一次； `WorkflowEngineClient` 和 `ExtensionSender` 各自拥有一个编排职责，委托通信工作。 diff --git a/docs/zh/DEVELOPER_GUIDE.md b/docs/zh/DEVELOPER_GUIDE.md index 5a65290c..8c4d7917 100644 --- a/docs/zh/DEVELOPER_GUIDE.md +++ b/docs/zh/DEVELOPER_GUIDE.md @@ -91,8 +91,10 @@ ExecutionResult result = ExecutePsop.builder() | `step_start` | 执行器 | 步骤开始 | `step` | | `task_request` | 执行器 | 子任务下发到 `onTask`/`onSelfTask` | `step`、`agent`、`task` | | `task_response` | 执行器 | `onTask`/`onSelfTask` 返回 `TaskResponse` | `step`、`agent`、`task`、`output` | +| `task_status_changed` | 执行器 | 任务状态变更（pending → running → success/failed） | `step`、`agent`、`task`、`status` | | `route_decision` | 执行器 | 分支选择 | `step`、`next`、`reason` | | `step_complete` | 执行器 | 步骤完成 | `step`、`results` | +| `workflow_complete` | 执行器 | 所有步骤完成 | `history`、`step_outputs` | | `agent_request` | 引擎客户端 | 消息发送到智能体 | `agent`、`request`、`metadata` | | `agent_response` | 引擎客户端 | 收到智能体响应 | `agent`、`response` | | `agent_status_update` | 引擎客户端 | 智能体 SSE 状态更新 | `agent`、`state`、`is_final` | @@ -200,11 +202,11 @@ AgentCard card = mapper.readValue(json, AgentCard.class); ```java WorkflowEngineClientConfig.builder() - .authProvider((agentName, agentCard, headers) -> { - headers.put("Authorization", "Bearer " + mySsoToken); - headers.put("X-Custom", "value"); - }) - .build(); + .authProvider((agentName, agentCard, headers) -> { + headers.put("Authorization", "Bearer " + mySsoToken); + headers.put("X-Custom", "value"); + }) + .build(); ``` ### 7.3 凭证文件字段 @@ -282,7 +284,7 @@ public Flux execute(@PathVariable String psopId) { ## 11. 检查清单 1. 添加 Maven 依赖 -2. 实现 `ControlPoint`（onTask + onSelfTask + onRoute） +2. 实现 `ControlPoint`（至少 onTask + onRoute；onSelfTask 和 onNegotiation 有默认实现） 3. 获取 AgentCards（从注册中心或 JSON 文件） 4. 加载 Workflow（通过 `LoadPsop` 或自行构建） 5. 配置 `.env` 和凭证文件 diff --git a/docs/zh/notification-t-design-analysis.md b/docs/zh/notification-t-design-analysis.md index cd767393..a384f839 100644 --- a/docs/zh/notification-t-design-analysis.md +++ b/docs/zh/notification-t-design-analysis.md @@ -59,7 +59,6 @@ Notification-T 和 Authorization-T 是 A2A-T 的前置扩展，在工作流启 Authorization-T 是一次性操作：下发策略 → 服务端确认 → 完成。不需要长连接。 流程： -``` ```mermaid graph LR ES["ExtensionSender.sendAuthorization"] @@ -70,7 +69,6 @@ graph LR ES --> TR --> SR --> ACK --> DONE ``` -``` ### 3.2 Notification-T @@ -78,7 +76,6 @@ Notification-T 需要长连接：订阅 → 流保持打开 → 后续结果通 设计选择：**在独立守护线程上维持长连接 SSE 流**。 -``` ```mermaid graph TD ES["ExtensionSender.sendNotification"] @@ -92,7 +89,6 @@ graph TD ES --> TR --> DT --> ACK --> FUT --> OPEN --> PUSH --> END ``` -``` --- diff --git "a/docs/zh/\345\267\245\344\275\234\345\217\260AgentScope\347\273\223\345\220\210\346\226\271\346\241\210.md" "b/docs/zh/\345\267\245\344\275\234\345\217\260AgentScope\347\273\223\345\220\210\346\226\271\346\241\210.md" index ea42407a..2205313b 100644 --- "a/docs/zh/\345\267\245\344\275\234\345\217\260AgentScope\347\273\223\345\220\210\346\226\271\346\241\210.md" +++ "b/docs/zh/\345\267\245\344\275\234\345\217\260AgentScope\347\273\223\345\220\210\346\226\271\346\241\210.md" @@ -53,26 +53,21 @@ A2A-T 执行引擎 SDK 和 AgentScope 解决的是完全不同层次的问题： ### 3.2 核心 API 表面 -``` -编排中心 (Python/FastAPI) - | PSOP 工作流存储 + 搜索 API - v -LoadPsop.search() / load() <- 从编排中心搜索 + 加载工作流 - | -ExecutePsop.builder() <- 高层工作流执行器（加载->执行->事件->持久化） - .psop(workflow) - .agentCards(cards) - .controlPoint(cp) <- 工作台业务决策回调（AgentScope 插入点） - .engineClient(client) - .execute() - | - +- WorkflowEngineClient <- 出站发送门面（Task-T + 协商循环 + 认证） - | +- sendMessage(agent, msg) - | - +- ExtensionSender <- 前置预定位门面（Authorization-T / Notification-T） - | +- prePosition() - | - +- A2ATransport <- 底层传输（共享，两门面复用） +```mermaid +graph TD + ORCH["编排中心 (Python/FastAPI)
PSOP 工作流存储 + 搜索 API"] + LOAD["LoadPsop.search() / load()
从编排中心搜索 + 加载工作流"] + EXEC["ExecutePsop.builder()
高层工作流执行器（加载->执行->事件->持久化）
.psop(workflow)
.agentCards(cards)
.controlPoint(cp) ← 工作台业务决策回调（AgentScope 插入点）
.engineClient(client)
.execute()"] + WEC["WorkflowEngineClient
出站发送门面（Task-T + 协商循环 + 认证）
sendMessage(agent, msg)"] + EXT["ExtensionSender
前置预定位门面（Authorization-T / Notification-T）
prePosition()"] + TRANSPORT["A2ATransport
底层传输（共享，两门面复用）"] + + ORCH --> LOAD + LOAD --> EXEC + EXEC --> WEC + EXEC --> EXT + WEC --> TRANSPORT + EXT --> TRANSPORT ``` ### 3.3 工作台需要实现的三个组件 diff --git a/pom.xml b/pom.xml index 6b8a941d..5cb24a24 100644 --- a/pom.xml +++ b/pom.xml @@ -36,9 +36,9 @@ - scm:git:git@github.com:Zhoujie628/a2at-engine-java.git - scm:git:git@github.com:Zhoujie628/a2at-engine-java.git - https://github.com/Zhoujie628/a2at-engine-java + scm:git:git@github.com:project-openan/workflow-engine-sdk-java.git + scm:git:git@github.com:project-openan/workflow-engine-sdk-java.git + https://github.com/project-openan/workflow-engine-sdk-java