开发实施技术方案 · DSH SDK 集成

DeepSeek Harness SDK 集成 · 开发实施技术方案

输入基线:架构设计方案.md(v1.1,已批准)· 目标:单机 Python 服务端 · 数百用户 × 每用户 1~2 并发会话 · 技能三来源(私有/共享/远程注册中心) 硬约束(用户确认):Python 3.12 锁定 · 元数据库支持多类型数据库(SQLite/PostgreSQL/MySQL)· 自带用户聊天前端 + 技能管理前端(贴近 dsh 客户端风格)· 最终以 Python 包(库 + CLI 双形态) 交付供其他项目引用 本文档回答"怎么开发":技术栈、工程结构、核心模块、数据模型、里程碑(M0–M3)、验收与验证、风险处置。


1. 目标与范围

1.1 目标

在单机 Python 服务端上落地一套"每用户独立 dsh 运行时"的多租户服务,并以 Python 包形态交付:

1.2 非本期范围(对齐架构 §1.2)

1.3 技术基线(锁定)

项 锁定值 说明
Python 3.12.x(requires-python = ">=3.12,<3.13") 全仓库(服务端 + 测试)统一
数据库 SQLite(默认/开发)· PostgreSQL · MySQL(可切换) 同一套 SQLAlchemy 模型,DATABASE_URL 驱动
前端 React 18 + Vite + TypeScript 自研,贴近 dsh 客户端布局
dsh 运行时 dsh@0.2.0-rc.2(锁定) 升级走灰度 + 协议回归

2. 技术栈与选型

层 选型 理由 / 备注
语言 Python 3.12(asyncio) 子进程生命周期与 NDJSON 流天然异步;uvicorn 单进程承载数百会话
Web 框架 FastAPI + uvicorn 异步友好、WebSocket 一等公民、OpenAPI 自动生成
数据库 SQLAlchemy 2.x(async)+ 多后端:SQLite / PostgreSQL / MySQL 模型与业务层方言无关;DATABASE_URL 一键切换;Alembic 多后端迁移
迁移 Alembic 每个数据库一个迁移链(同一套 upgrade 脚本,方言差异收敛在模型/迁移内)
认证 JWT(pyjwt)+ 自研 RBAC 中间件 贴合架构 §4.2 认证/授权边界;前端登录态由 HttpOnly cookie 或 Authorization 头承载(可配)
子进程 asyncio.subprocess + 自研 NDJSON 帧解析 SDK 协议仅 3 个请求方法 + 4 类通知,无需第三方库
配置 pydantic-settings + YAML(.env 覆盖) profile patch 渲染用 PyYAML
技能 manifest PyYAML 校验 frontmatter 对齐架构 §7.4
前端构建 Vite + React + TS;构建产物打进 wheel(package_data) 库模式/CLI 模式都随包自带前端,无需单独部署前端
打包 hatchling(或 poetry)→ wheel 发布 PyPI/私有索引 [project.scripts] dsh-server = "dsh_server.cli:main"
测试 pytest + pytest-asyncio;集成测试直接拉起真实 dsh 二进制 协议回归必须跑真二进制;多数据库 CI 矩阵
压测 locust 或自编 asyncio 压测脚本 M3 验收用
进程运行账号 起步共享运行账号 + 每用户目录 ACL;cgroup v2 按用户分片(可选) 架构 §8 备注

多数据库策略:


3. 工程结构(monorepo,一个 wheel 交付)

dsh-python-integration/
├── 架构设计方案.md
├── 开发实施技术方案.md           # 本文档
├── server/                        # Python 服务端(打包为一个 wheel:dsh-harness-server)
│   ├── pyproject.toml             # hatchling;requires-python >=3.12,<3.13;[project.scripts] dsh-server
│   ├── .env.example
│   └── src/dsh_server/
│       ├── __init__.py            # 版本号(打包/API 统一从这取)
│       ├── cli.py                 # CLI 入口:dsh-server serve / init-db / create-user / skill-sync
│       ├── embed.py               # 库模式入口:create_app(settings) -> FastAPI;start_runtime_pool()
│       ├── config.py              # pydantic-settings:DSH 路径、沙箱模式、回收参数、DATABASE_URL
│       ├── api/
│       │   ├── router.py          # 汇总挂载:/api/v1 + WS
│       │   ├── sessions.py        # POST/GET/DELETE /api/v1/sessions
│       │   ├── prompts.py         # POST /api/v1/sessions/{sid}/prompt
│       │   ├── events.py          # WS /api/v1/sessions/{sid}/events
│       │   ├── results.py         # GET /api/v1/sessions/{sid}/result
│       │   ├── skills.py          # GET/POST /api/v1/skills(目录/发布/授权)
│       │   ├── quota.py           # GET /api/v1/quota
│       │   └── auth_routes.py     # POST /api/v1/auth/token(登录换 JWT)
│       ├── auth/
│       │   ├── jwt.py             # 签发/校验 JWT
│       │   └── rbac.py            # 角色解析 + 资源授权装饰器(依赖注入而非全局可变)
│       ├── sdkclient/
│       │   ├── transport.py       # NDJSON 帧读写 + 行缓冲 + 超时
│       │   ├── client.py          # initialize/session_prompt/shutdown + 通知分发
│       │   └── errors.py          # DshProtocolError / DshTimeout / DshCrash
│       ├── orchestrator/
│       │   ├── runtime.py         # RuntimeHandle:单用户进程句柄/健康/世代
│       │   ├── manager.py         # RuntimeManager:uid→RuntimeHandle 注册表
│       │   ├── lifecycle.py       # 状态机 absent→provisioning→ready→…(§5.2)
│       │   ├── recycle.py         # 空闲扫描 + shutdown→SIGTERM→SIGKILL
│       │   └── quotas.py          # 并发/token/进程数配额(应用层计数)
│       ├── events/
│       │   ├── bus.py             # asyncio.Queue 内部事件总线(订阅者:WS 推送/结果桥接/审计)
│       │   ├── normalize.py       # JSON-RPC 通知 → 内部统一事件(架构 §7.3)
│       │   └── result_bridge.py   # PromptTracker:messageId → 事件区间 → idle 汇总结果
│       ├── skills/
│       │   ├── manifest.py        # SKILL.md frontmatter 解析/校验(架构 §7.4)
│       │   ├── sync.py            # B1:远程→物化目录→原子切 current 软链
│       │   ├── grants.py          # SkillGrant 模型 + 按角色过滤
│       │   └── service.py         # 技能目录 API 逻辑 + 审计写库
│       ├── profiles/
│       │   └── render.py          # 按用户渲染 cordis.patch.yml(§5.4 配置点)
│       ├── db/
│       │   ├── base.py            # DeclarativeBase + async engine factory(按 DATABASE_URL)
│       │   ├── models.py          # SQLAlchemy 模型(§6,方言无关)
│       │   └── migrations/        # Alembic(env.py 按 dialect 分支;versions/ 多后端共用迁移脚本)
│       └── webassets/             # 前端构建产物(vite build 输出,随 wheel 打进 package_data)
├── web/                           # 自研 React 前端(源码):聊天页 + 技能管理页
│   ├── package.json               # React 18 + Vite + TS
│   ├── vite.config.ts             # build.outDir -> ../server/src/dsh_server/webassets
│   ├── src/
│   │   ├── chat/                  # 用户聊天页(贴近 dsh 布局:左会话栏 + 中对话流 + 底部输入)
│   │   ├── skills/                # 技能管理页(目录/发布/授权/回滚)
│   │   ├── api/                   # fetch + WS 客户端(同源 /api/v1)
│   │   └── theme/                 # dsh 风格主题(明/暗,紧凑对话密度)
│   └── tests/                     # Vitest + Playwright(e2e)
├── tests/                         # Python 测试(server 包测试与前端 e2e 分流)
│   ├── unit/
│   ├── integration/               # 真实 dsh 二进制端到端
│   └── conftest.py
└── scripts/
    ├── dev-up.sh                  # 同时起 uvicorn + vite dev
    ├── skill-sync.sh              # 手动触发 B1 物化
    └── smoke.sh                   # curl 冒烟:登录→建会话→prompt→result→删除

库模式可用性的硬性要求:dsh_server 内不得有 import 时副作用(不读 .env、不起进程、不连库)——所有初始化走 create_app(settings) / start_runtime_pool(settings) 显式传入;这样别的项目 pip install dsh-harness-server 后 import 无副作用,可随时启动。


4. 核心模块设计要点

4.1 sdkclient/transport.py — NDJSON 帧

class SdkClient:
    async def initialize(self, model: str, cwd: Path, max_tokens: int | None = None) -> ServerInfo: ...
    async def session_prompt(self, session_id: str, content: str) -> MessageId: ...   # 仅准入回执
    async def shutdown(self) -> None: ...
    # 通知回调由构造时注入:on_event / on_status / on_subagent

4.2 orchestrator/lifecycle.py — 每用户状态机

absent → provisioning(拉起+initialize) → ready(可接会话)
                       │                       │
                       │ 崩溃                空闲超时 / 显式关闭 / 配额
                       ▼                       ▼
                restarting(退避)         shutting_down(shutdown→SIGTERM→SIGKILL)
                       │                       │
                       └───────── back to ready / removed

4.3 events/result_bridge.py — 事件→结果桥接

4.4 skills/sync.py — B1 目录同步 + 原子切换

4.5 skills/manifest.py — SKILL.md 校验

4.6 auth/rbac.py — 权限

4.7 前端(自研 React,贴近 dsh 客户端)


5. 里程碑与任务分解

后端 M0–M3 对齐架构 P0–P3;前端任务穿插其中(M1 聊天页、M2 技能管理页),包化在 M3 收口;每个里程碑"可运行 + 可验收"。

M0 — 工程骨架 + 单用户端到端 + CLI 雏形(≈ P0,1–2 周)

# 任务 文件 验收
M0.1 工程脚手架:pyproject(3.12 锁定、dsh-server 入口)、config、DB 初始化(SQLite + Alembic 首版)、.env.example server/ uv run pytest 绿;dsh-server --help 可用
M0.2 sdkclient:transport + client(握手/超时/崩溃) sdkclient/* 单测帧解析/超时;集成:spawn 真 dsh → initialize 返回 serverInfo
M0.3 API 骨架 + JWT 登录 + RBAC 中间件 api/*, auth/* curl:401 无 token;登录换 token;建会话返回 u:<uid>:<uuid>
M0.4 编排器最小状态机 + 事件总线 + 规范化 + 结果桥接 + WS orchestrator/*, events/* 集成:建会话→prompt→事件→idle→result→删除闭环
M0 验收 scripts/smoke.sh 全绿;test_e2e.py 过 架构 §9 P0:端到端一个会话闭环

M1 — 多用户隔离 + 聊天前端 V1(≈ P1,2–3 周)

# 任务 文件 验收
M1.1 每用户 DSH_HOME/workspace 布局 + ACL orchestrator/runtime.py 双用户目录互不相交(集成断言)
M1.2 配额(并发/进程/token) orchestrator/quotas.py 超配额 429/409
M1.3 崩溃重启 + 退避熔断;空闲回收 + SIGTERM/SIGKILL lifecycle.py, recycle.py kill -9 → 自动重启;空闲超时 → 优雅回收
M1.4 审计日志;stderr 按用户落盘;stdout 纯净守护 events/normalize.py, runtime.py 审计表有记录;stdout 每帧可解析(无脏行)
M1.5 聊天前端 V1:React 骨架 + 主题 + 会话栏/对话流/输入框 + WS 实时 web/src/chat/* Playwright:登录→建会话→发消息→渲染事件→得结果
M1 验收 双用户互不可见互不影响;故障注入全过;聊天页可用 架构 §9 P1 全项 + 前端可用

M2 — 技能三来源 + RBAC + 技能管理前端 + 多数据库接入(≈ P2,2–3 周)

# 任务 文件 验收
M2.1 技能目录 API + manifest 校验 + 版本管理 skills/manifest.py, service.py 发布校验失败拒绝;目录可查
M2.2 B1 同步 + 原子软链切换 + 回滚 skills/sync.py, scripts/skill-sync.sh 发布 v2 → current 切 v2;回滚 <1s
M2.3 用户私有技能写路径(仅本人) api/skills.py A 私有技能仅 A 可写
M2.4 RBAC:SkillGrant + 按角色过滤 skills/grants.py, auth/rbac.py engineer 看不到 admin 专属技能
M2.5 技能管理前端:目录/发布/授权/回滚页 web/src/skills/* Playwright:admin 发布→回滚→授权全场
M2.6 多数据库接入:engine factory 按 DATABASE_URL;Alembic 多后端;CI 三方言矩阵 db/base.py, migrations/* DATABASE_URL=postgresql://… / mysql://… 下全量 pytest 绿
M2 验收 三来源并集可用;角色隔离;热更新无重启;回滚秒级;三种数据库跑通 架构 §9 P2 全项 + 多库矩阵

M3 — 规模、包化发布与进阶(≈ P3,3–4 周)

# 任务 文件 验收
M3.1 压测 300 用户 × 1–2 会话 tests + locust P95 建会话 <2s;OOM 只伤单用户
M3.2 cgroup v2 按用户分片(可选) quotas.py + 部署脚本 超限被内核限制
M3.3 OTel 埋点 events/bus 薄封装 关键链路可见
M3.4 包化收口:webassets 打进 wheel;embed.py 完整(create_app/start_runtime_pool);PyPI/私有索引发布 pyproject.toml, embed.py, CI 干净 venv:pip install dsh-harness-server → dsh-server serve 起服务 smoke 过;from dsh_server.embed import create_app 嵌入示例项目跑通
M3.5 (可选)B2 注册中心直连插件;出网白名单 skills/ 权限实时;失败回退 B1
M3 验收 压测达标;wheel 双形态可用;文档(README/快速上手/嵌入示例)齐 架构 §9 P3 验收 + 包交付

6. 数据模型(SQLAlchemy 方言无关)

# 关键模型(简写;全部用方言无关类型)
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    uid: Mapped[str] = mapped_column(String(64), unique=True)      # 对外 uid
    display_name: Mapped[str | None] = mapped_column(String(128))
    roles: Mapped[list] = mapped_column(JSON, default=list)        # ["engineer", ...]

class Session(Base):
    __tablename__ = "sessions"
    id: Mapped[str] = mapped_column(String(96), primary_key=True)  # u:<uid>:<uuid>
    uid: Mapped[str] = mapped_column(ForeignKey("users.uid"))
    model: Mapped[str] = mapped_column(String(128))
    cwd: Mapped[str] = mapped_column(String(512))
    status: Mapped[str] = mapped_column(String(24), default="ready")
    runtime_generation: Mapped[int] = mapped_column(Integer, default=0)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
    last_active_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

class Skill(Base):        # 目录;/ skill_versions(版本+物化路径)/ skill_grants(role, skill_id, version_constraint, enabled)
    __tablename__ = "skills"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(128), unique=True)
    description: Mapped[str] = mapped_column(Text)
    latest_version: Mapped[str] = mapped_column(String(32))
    disabled: Mapped[bool] = mapped_column(Boolean, default=False)

class QuotaLedger(Base): ...   # uid, session_id, tokens, at
class AuditLog(Base): ...      # at, actor, action, detail(JSON)

7. 验证与质量策略

层 手段 覆盖
单元测试 pytest + pytest-asyncio(帧解析、状态机、manifest、grants、result_bridge、RBAC、多数据库 CRUD) 纯逻辑快路径
集成测试 真实 dsh 二进制,临时 DSH_HOME/workspace fixture 协议回归、stdout 纯净性、崩溃恢复
API 冒烟 scripts/smoke.sh(登录→会话→prompt→result→删除) 部署后 30 秒健康检查
数据库矩阵 CI:test-{sqlite,postgres,mysql} 三个 job(复用同一套 pytest) 方言兼容
前端 Vitest(组件)+ Playwright(e2e:登录/会话/技能管理) 聊天页与技能管理页关键流
故障注入 kill -9 / 写满配额 / 断模型网络(mock) 崩溃重启、熔断、配额、超时
压测 locust:300 用户并发建会话/提示词 P95 延迟、内存上限、回收稳定性
包验证 CI 步骤:干净 venv pip install dist/*.whl → dsh-server serve smoke + 嵌入示例 wheel 一致性

8. 兼容性边界与退休路径


9. 风险与处置(增量于架构 §10)

风险 处置
多数据库方言差异(迁移/类型/并发) 模型层仅方言无关类型;CI 三库矩阵;SQLite 仅限开发/小站
Python 3.12 生态兼容(asyncpg/asyncmy 版本) CI 锁 3.12 全依赖矩阵;装 wheel 前跑最小 smoke
前端"贴近 dsh"还原度 以信息架构/布局/密度对齐为准,不逐像素复制;主题可配置化
库模式误用(import 副作用/全局状态) embed 化设计:所有状态经 create_app(settings) 注入;测试覆盖"import 无副作用"
SQLite 单写者限制(多 worker) worker=1 直到 M3;切 PG 后可多 worker
数百子进程 fd 消耗 asyncio 无线程 + 每用户 1 进程;监控 fd 上限
空闲回收误杀进行中轮次 回收前检查 turnNeedsResult 非空则顺延
WS 积压 事件总线背压(队列上限 + drop-oldest;结果桥接独立于推送不受损)
skill-sync 与热发现竞争 软链原子替换,无中间态

10. 开工清单(第一个开发者 Day-1)

  1. cp server/.env.example server/.env:配 DSH_BIN、DATABASE_URL=sqlite:///dev.db、JWT secret、回收参数;
  2. cd server && uv sync(Python 3.12),uv run alembic upgrade head;
  3. bash scripts/dev-up.sh(后端 + vite dev 同起);
  4. 先读 tests/conftest.py(临时 DSH_HOME fixture)→ 跑 M0.2 的 sdkclient 集成测试,确认本机 dsh 可 spawn;
  5. 按 M0 → M1 → M2 → M3 推进,每步跑对应验收;M2 起把 DATABASE_URL 切换到 PG/MySQL 各跑一遍矩阵。

附:与架构文档的映射

本方案章节 架构文档章节
§2 技术栈/多数据库 §8(容量/部署)+ 摘要决策 5
§3 工程结构/包形态 §3(总体架构分层)
§4.1–4.3 模块 §2.2 协议 / §7.2 状态机 / §6 时序
§4.4–4.5 技能 §5(三来源/B1/B2/manifest)
§4.6 RBAC §4.2/§5.3
§4.7 前端 §1.1 技术基线(自研 React 贴近 dsh)
§5 里程碑 §9 演进路线(P0–P3)
§6 数据模型 §7.1/§7.3

DeepSeek Harness SDK 集成 · 发布于 Cloudflare Pages · 由设计文档自动生成