前言

从本文开始，我们将进入一个完整的实战案例——“鸿蒙壁纸大师”。这是一个基于 A2A（Agent-to-Agent）协议构建的生产级智能体应用，用户通过与小艺对话，即可生成各种风格的 AI 壁纸。

本项目代码托管在：ai-wallpaper-server，本文所有代码均来源于实际项目，可直接运行。

本系列将分 10 篇文章，从架构设计到代码实现，完整展示一个鸿蒙 A2A 智能体的全流程开发。本文是第一篇：项目概述与架构设计。

一、项目背景

1.1 为什么选择 A2A 模式

在鸿蒙智能体平台中，开发者可以选择多种编排方式（详见本系列第 1 篇）。对于壁纸生成这类需要复杂业务逻辑和自有后端服务的场景，A2A 模式是最佳选择：

模式 适用场景 为什么不选
单 Agent 纯对话场景 需要调用外部 AI 模型，单 Agent 无法满足
工作流 平台内部编排 需要自定义 API 服务和模型调用
OpenClaw 第三方 API 接入 需要复杂的业务逻辑和状态管理
A2A 自有服务集成 ✅ 完全自控，灵活扩展

1.2 核心能力

鸿蒙壁纸大师具备以下核心能力：

┌─────────────────────────────────────────────┐
│              鸿蒙壁纸大师                      │
├─────────────────────────────────────────────┤
│  ● 多轮交互：选择风格 → 选择主题 → 选择场景    │
│  ● AI 生成：调用火山引擎文生图模型生成壁纸      │
│  ● 智能优化：LLM 优化提示词，提升壁纸质量       │
│  ● 设备适配：根据设备型号自动优化壁纸尺寸       │
│  ● 异步推送：生成完成后通过 Webhook 通知用户    │
│  ● 实时进度：SSE 流式推送生成进度               │
│  ● 卡片展示：通过 DisplayFaCard 展示壁纸       │
└─────────────────────────────────────────────┘

鸿蒙壁纸大师 A2A 架构示意图

上图展示了鸿蒙壁纸大师的整体架构：用户通过小艺触发 A2A 请求，壁纸服务器处理后调用火山引擎模型生成壁纸

二、整体架构

2.1 系统架构图

┌──────────────┐     A2A协议      ┌──────────────────┐
│  用户设备     │ ◄────────────►  │  小艺开放平台     │
│  (鸿蒙手机)   │                 │  (智能体执行引擎)  │
└──────────────┘                 └────────┬─────────┘
                                          │ A2A/JSON-RPC
                                          ▼
┌────────────────────────────────────────────────────────┐
│              AI Wallpaper Server (FastAPI)              │
│                                                        │
│  ┌──────────┐  ┌──────────┐  ┌──────────────────────┐ │
│  │  Routes   │  │ Services │  │     Models/Schemas   │ │
│  │ 路由层    │  │ 服务层   │  │     数据模型         │ │
│  ├──────────┤  ├──────────┤  ├──────────────────────┤ │
│  │ agent_   │  │ wallpaper│  │ JsonRpcRequest/      │ │
│  │ routes.py│  │ _service │  │ Response             │ │
│  │          │  │ intent_  │  │ CardData/CardInfo    │ │
│  │          │  │ service  │  │ A2ATask/Message      │ │
│  │          │  │ wallpaper│  │                      │ │
│  │          │  │ _flow_   │  │                      │ │
│  │          │  │ service  │  │                      │ │
│  │          │  │ webhook_ │  │                      │ │
│  │          │  │ push_    │  │                      │ │
│  └──────────┘  └──────────┘  └──────────────────────┘ │
│                                                        │
│  ┌──────────┐  ┌──────────┐  ┌──────────────────────┐ │
│  │  Utils   │  │  Config  │  │      Core            │ │
│  ├──────────┤  ├──────────┤  ├──────────────────────┤ │
│  │ intent_  │  │ device_  │  │ config.py (Settings) │ │
│  │classifier│  │resolution│  │ .env 环境变量        │ │
│  │ logger   │  │          │  │                      │ │
│  └──────────┘  └──────────┘  └──────────────────────┘ │
└──────────────────────────┬─────────────────────────────┘
                           │ HTTP/HTTPS
                           ▼
┌────────────────────────────────────────────────────────┐
│                    火山引擎 API                          │
│                                                        │
│  ┌──────────────────┐  ┌─────────────────────────────┐ │
│  │ Language Model   │  │  Image Generation Model     │ │
│  │ doubao-pro       │  │  doubao-seedream-5-0       │ │
│  │ - 意图识别       │  │  - 文生图                   │ │
│  │ - 提示词优化     │  │  - 根据 prompt 生成壁纸     │ │
│  │ - 闲聊回复       │  │                             │ │
│  └──────────────────┘  └─────────────────────────────┘ │
└────────────────────────────────────────────────────────┘

2.2 技术栈选择

组件 技术选型 选择理由
Web 框架 FastAPI 高性能异步框架，原生支持 SSE
HTTP 客户端 httpx 异步 HTTP 请求，支持流式响应
数据验证 Pydantic v2 类型安全的数据模型
A2A 协议 兼容 Google A2A + 鸿蒙扩展 标准化的 Agent 通信协议
AI 模型 火山引擎 (豆包) 兼具语言理解和文生图能力
部署方式 Docker + systemd 生产级容器化部署

2.3 核心数据流

一次完整的壁纸生成请求涉及以下数据流转：

"""
一次完整的壁纸生成请求数据流：

Step 1: 用户 → 小艺
    用户："帮我生成一张星空壁纸"

Step 2: 小艺 → 智能体执行引擎
    → 意图识别：generate_wallpaper
    → 选择智能体：壁纸大师

Step 3: 智能体执行引擎 → AI Wallpaper Server (A2A)
    POST /agent/message
    {
        "jsonrpc": "2.0",
        "method": "message/stream",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"kind": "text", "text": "帮我生成一张星空壁纸"}]
            }
        }
    }

Step 4: AI Wallpaper Server → 火山引擎 (LLM)
    POST /chat/completions
    → 意图识别：用户想生成壁纸
    → 返回风格选项卡片

Step 5: 用户选择风格 → ... (多轮交互)

Step 6: 最终生成 → 调用火山引擎文生图
    POST /images/generations
    {
        "model": "doubao-seedream-5-0",
        "prompt": "...优化后的提示词...",
        "size": "1080x2400"
    }

Step 7: AI Wallpaper Server → 小艺 (SSE 流式返回)
    data: {"result": {"kind": "status-update", ...}}
    data: {"result": {"kind": "artifact-update", ...}}

Step 8: 小艺 → 用户
    显示生成的壁纸卡片
"""

三、项目目录结构

3.1 完整的目录结构

ai-wallpaper-server/
├── main.py                    # 应用入口，FastAPI 应用初始化
├── requirements.txt           # Python 依赖
├── Dockerfile                 # Docker 构建文件
├── docker-compose.yml         # Docker Compose 编排
├── .env.example               # 环境变量模板
│
├── core/
│   ├── __init__.py
│   └── config.py              # 配置管理 Settings 类
│
├── config/
│   ├── __init__.py
│   └── device_resolution.py   # 设备分辨率映射配置
│
├── models/
│   ├── __init__.py
│   └── schemas.py             # JSON-RPC / A2A / 卡片数据模型
│
├── routes/
│   ├── __init__.py
│   └── agent_routes.py        # A2A 路由处理（核心，1559行）
│
├── services/
│   ├── __init__.py
│   ├── wallpaper_service.py   # 壁纸生成服务（LLM+文生图）
│   ├── wallpaper_flow_service.py  # 壁纸流程管理器
│   ├── intent_service.py      # 意图识别服务
│   └── webhook_push_service.py    # Webhook 推送服务
│
├── utils/
│   ├── __init__.py
│   ├── intent_classifier.py   # 意图分类器
│   └── logger.py              # 日志配置（多文件分类）
│
└── tests/
    ├── __init__.py
    ├── test_api.py             # API 测试
    ├── test_intent.py          # 意图识别测试
    ├── test_intent_llm.py      # LLM 意图识别测试
    ├── test_refactor.py        # 重构测试
    ├── manual_test.py          # 手动测试脚本
    └── integration_test.py     # 集成测试

3.2 各模块职责

模块 文件 职责 核心功能
入口 main.py FastAPI 应用创建与生命周期 路由注册、Session/上下文/任务状态管理
配置 core/config.py 全局配置 Settings 类、环境变量读取
设备 config/device_resolution.py 设备分辨率映射 手机/平板/折叠屏的分辨率配置
数据模型 models/schemas.py 协议数据结构 JSON-RPC、A2A、卡片模型
路由 routes/agent_routes.py A2A 协议路由 7 种 RPC 方法处理（核心）
壁纸服务 services/wallpaper_service.py 壁纸生成 LLM 调用、文生图调用、SSE 流式推送
流程管理 services/wallpaper_flow_service.py 多轮交互流程 风格/主题/场景三步骤状态机
意图识别 utils/intent_classifier.py 用户意图分类 7 种意图类型的 LLM 分类器
Webhook services/webhook_push_service.py 异步消息推送 HMAC-SHA256 签名推送
日志 utils/logger.py 多文件日志 请求/SSE/火山引擎/通用日志分离

四、应用入口详解

4.1 main.py 核心代码

"""
AI Wallpaper API Service
基于鸿蒙 Agent 通信协议（兼容 Google A2A）实现的 API 服务
采用 Streamable HTTP + JSON RPC 协议，支持 SSE 流式输出
转发请求到火山引擎 OpenAI 兼容接口
"""
import json
from datetime import datetime, timedelta
from typing import Dict, List, Any

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from utils.logger import logger, set_session_id
from core.config import settings
from routes.agent_routes import router as agent_router
from services.wallpaper_flow_service import flow_managers


# ============== 内存存储（生产环境建议使用 Redis）=============

agent_sessions: Dict[str, Dict[str, Any]] = {}       # Session 存储
conversation_contexts: Dict[str, List[Dict]] = {}     # 对话上下文
task_states: Dict[str, Dict[str, Any]] = {}           # 任务状态


async def lifespan(app: FastAPI):
    """FastAPI lifespan 事件处理器"""
    # 启动事件
    logger.info("=" * 60)
    logger.info("AI Wallpaper API 启动")
    logger.info(f"VOLCANIC_BASE_URL: {settings.VOLCANIC_BASE_URL}")
    logger.info(f"LANGUAGE_MODEL: {settings.LANGUAGE_MODEL}")
    logger.info(f"IMAGE_MODEL: {settings.IMAGE_MODEL}")
    logger.info(f"INTENT_MODEL: {settings.INTENT_MODEL}")
    logger.info("=" * 60)

    # 将共享状态注入到 app.state
    app.state.agent_sessions = agent_sessions
    app.state.conversation_contexts = conversation_contexts
    app.state.task_states = task_states
    yield
    # 停止事件
    logger.info("AI Wallpaper API 关闭")


# 创建 FastAPI 应用
app = FastAPI(
    title="AI Wallpaper API",
    description="基于鸿蒙 Agent 通信协议的 AI 壁纸生成 API 服务",
    version="1.0.0",
    lifespan=lifespan,
)

app.include_router(agent_router)  # 注册路由

4.2 关键 API 端点

端点 方法 用途 说明
/ GET 服务信息 返回服务名称、版本、协议等信息
/health GET 健康检查 用于 Docker 健康检查和监控
/agent/message POST A2A 消息处理 所有 A2A 协议的 RPC 方法入口
/agent/ GET Agent 信息 返回 Agent 名称、端点、模型
/webhook/push POST Webhook 回调 接收平台的推送通知

4.3 模型配置

# core/config.py - 配置管理

class Settings:
    """应用配置"""

    def __init__(self):
        # 服务配置
        self.HOST = os.getenv("HOST", "0.0.0.0")
        self.PORT = int(os.getenv("PORT", "8080"))

        # 火山引擎配置
        self.VOLCANIC_API_KEY = os.getenv("VOLCANIC_API_KEY", "")
        self.VOLCANIC_BASE_URL = os.getenv(
            "VOLCANIC_BASE_URL",
            "https://ark.cn-beijing.volces.com/api/v3"
        )

        # 三种不同用途的模型
        self.LANGUAGE_MODEL = os.getenv("LANGUAGE_MODEL", "doubao-seed-2-0-mini-250115")
        self.INTENT_MODEL = os.getenv("INTENT_MODEL", "doubao-lite-4k")
        self.IMAGE_MODEL = os.getenv("IMAGE_MODEL", "doubao-seedream-5-0-260128")

        # Webhook 推送配置
        self.WEBHOOK_API_KEY = os.getenv("WEBHOOK_API_KEY", "")
        self.WEBHOOK_API_SECRET = os.getenv("WEBHOOK_API_SECRET", "")
        self.WEBHOOK_API_ID = os.getenv("WEBHOOK_API_ID", "")

五、A2A 协议实现

5.1 支持的 RPC 方法

本项目实现了鸿蒙 A2A 协议的 7 种 RPC 方法：

方法 用途 是否必须
initialize 会话初始化 ✅
notifications/initialized 初始化通知 ✅
message/stream 流式消息处理 ✅
tasks/cancel 取消任务 ✅
clearContext 清理上下文 ❌
authorize 用户授权 ❌
deauthorize 解授权 ❌

5.2 路由分发逻辑

# routes/agent_routes.py - 路由入口

@router.post("/message")
async def handle_agent_message(
    request: Request,
    agent_session_id: Optional[str] = Header(None, alias="agent-session-id"),
):
    """统一的 Agent 消息处理接口"""

    body = await request.json()
    jsonrpc_request = JsonRpcRequest(**body)
    method = jsonrpc_request.method

    # 路由到不同的处理方法
    if method == "initialize":
        result = await handle_initialize(...)
    elif method == "notifications/initialized":
        result = await handle_initialized(...)
    elif method == "message/stream":
        return await handle_message_stream(...)  # SSE 流式响应
    elif method == "tasks/cancel":
        result = await handle_tasks_cancel(...)
    elif method == "clearContext":
        result = await handle_clear_context(...)
    elif method == "authorize":
        result = await handle_authorize(...)
    elif method == "deauthorize":
        result = await handle_deauthorize(...)
    else:
        result = JSONResponse(status_code=400, content=...)

    return result

六、为什么这个项目值得学习

6.1 生产级设计

本项目不是一个"玩具"项目，而是按照生产标准设计的：

  • 完善的日志系统：5 种分类日志（通用/请求/SSE/火山引擎）
  • 完整的错误处理：超时、重试、降级策略
  • Docker 部署：容器化 + 健康检查 + Compose 编排
  • 状态管理：会话、上下文、任务三种状态独立管理

6.2 实战参考价值

学习点 说明 对应文章
A2A 协议实现 7 种 RPC 方法的完整实现 第 30-32 篇
多轮交互设计 状态机驱动的流程管理 第 33 篇
LLM 集成 意图识别 + 提示词优化 + 文生图 第 34-35 篇
SSE 流式推送 实时进度更新 第 36 篇
Webhook 异步推送 生成完成通知 第 37 篇
设备适配 根据型号优化壁纸尺寸 第 38 篇
卡片交互 selectList + DisplayFaCard 第 39 篇

七、快速开始

如果你已经迫不及待想体验项目，可以按照以下命令快速启动：

# 克隆项目（示例仓库）
git clone https://github.com/example/ai-wallpaper-server.git
cd ai-wallpaper-server

# 复制环境变量模板
cp .env.example .env

# 安装依赖并启动服务
pip install -r requirements.txt
uvicorn main:app --host 0.0.0.0 --port 8080 --reload

快速开始仅用于本地体验，生产部署请参考后续文章。

总结

本文介绍了"鸿蒙壁纸大师"项目的整体架构和设计思路：

  1. 项目背景：选择 A2A 模式的原因
  2. 系统架构：三层架构（小艺平台 + 壁纸服务器 + 火山引擎）
  3. 目录结构：6 大模块的职责划分
  4. 应用入口：FastAPI 应用初始化和配置管理
  5. A2A 协议：7 种 RPC 方法的分发逻辑
  6. 学习价值：生产级项目的参考意义

接下来的文章将逐一深入每个模块的具体实现。下一篇我们将介绍环境搭建与基础配置，从零开始搭建开发环境。

如果这篇文章对你有帮助，欢迎点赞 👍、收藏 ⭐ 和关注 🎯！你的支持是我持续创作的动力！

相关资源

# harmonyos # 华为 # 算法 # 鸿蒙 # AI
Logo
人工智能6S服务平台

作为“人工智能6S店”的官方数字引擎，为AI开发者与企业提供一个覆盖软硬件全栈、一站式门户。

更多推荐

cover

鸿蒙实战｜分布式软总线与跨端协同：HarmonyOS设备协作体系深度剖析

avatar 人工智能6S服务平台
cover

鸿蒙三方库 | harmony-utils之WindowUtil屏幕亮度与常亮详解

avatar 人工智能6S服务平台
cover

鸿蒙ArkUI实战：用派生数据组合任务筛选与关键词搜索

avatar 人工智能6S服务平台