docs: Phase1 里程碑1 实施计划(骨架+数据模型+config 加载)
This commit is contained in:
@@ -0,0 +1,804 @@
|
||||
# Phase1 里程碑1(骨架+数据模型+config)实施计划
|
||||
|
||||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
**Goal:** 建立可安装的 src 布局项目骨架,实现 design §3+§9.4 全部业务数据模型与 config-design §1/§7 的配置加载,pytest 全绿。
|
||||
|
||||
**Architecture:** 标准 src 布局(`src/genesis/`)+ pyproject setuptools 打包 + pytest。数据模型一个模块统一集中(future.annotations 支持前向引用);config 用 pydantic 模型对应三 yaml,from_dir 加载并实现「env(yaml) 环境变量 > yaml > 默认值」优先级与敏感字段脱敏。
|
||||
|
||||
**Tech Stack:** Python ≥3.11(实际 3.14.3)、pydantic 2.13 / pydantic-settings 2.15、pyyaml、pytest 8。
|
||||
|
||||
## Global Constraints
|
||||
|
||||
- 全部文件修改遵循 `docs/superpowers/specs/2026-08-08-phase1-foundation-design.md`。
|
||||
- 数据模型文件 `src/genesis/data_models.py` 顶部必须有 `from __future__ import annotations`(design §9.4.6,P0-1)。
|
||||
- 字段名/默认值必须与 design.md §3 / §9.4 一一对应,不得增删。
|
||||
- 配置优先级:环境变量(`GENESIS_` 前缀 + `__` 嵌套分隔)> yaml 文件 > pydantic 默认值。
|
||||
- 敏感键(含 `key`/`secret`/`token` 的字段)在 `get_redacted()` 输出为 `"***"`。
|
||||
- 交流语言统一中文(代码注释、提交信息用中文;标识符/技术术语英文)。
|
||||
- 每个任务结束前运行 `pytest` 全绿,随后 commit。
|
||||
|
||||
---
|
||||
|
||||
### Task 1: 项目骨架 + pyproject 打包
|
||||
|
||||
**Files:**
|
||||
- Create: `pyproject.toml`
|
||||
- Create: `src/genesis/__init__.py`
|
||||
- Create: `tests/__init__.py`
|
||||
- Create: `tests/test_smoke.py`
|
||||
|
||||
**Interfaces:**
|
||||
- Consumes: 无(首个任务)
|
||||
- Produces: 包 `genesis`(`import genesis` 可用,`genesis.__version__ == "0.1.0"`);pytest 可从根目录运行
|
||||
|
||||
- [ ] **Step 1: 写失败测试(冒烟)**
|
||||
|
||||
`tests/test_smoke.py`:
|
||||
```python
|
||||
import genesis
|
||||
|
||||
|
||||
def test_package_importable():
|
||||
assert genesis.__version__ == "0.1.0"
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 写骨架文件**
|
||||
|
||||
`src/genesis/__init__.py`:
|
||||
```python
|
||||
"""Genesis:概要设计书自动生成 Agent。"""
|
||||
__version__ = "0.1.0"
|
||||
```
|
||||
|
||||
`tests/__init__.py`: 空文件。
|
||||
|
||||
`pyproject.toml`:
|
||||
```toml
|
||||
[build-system]
|
||||
requires = ["setuptools>=69"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "genesis"
|
||||
version = "0.1.0"
|
||||
description = "概要设计书自动生成 Agent"
|
||||
requires-python = ">=3.11"
|
||||
dependencies = [
|
||||
"pydantic>=2.13",
|
||||
"pydantic-settings>=2.15",
|
||||
"pyyaml>=6.0",
|
||||
"openpyxl>=3.1",
|
||||
"python-docx>=1.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = ["pytest>=8.0"]
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["src"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
```
|
||||
|
||||
- [ ] **Step 3: 运行测试,确认失败**
|
||||
|
||||
Run: `python -m pytest tests/test_smoke.py -v`
|
||||
Expected: FAIL(`ModuleNotFoundError: No module named 'genesis'`)
|
||||
|
||||
- [ ] **Step 4: 可编辑安装 + 运行测试,确认通过**
|
||||
|
||||
Run: `python -m pip install -e ".[dev]" -i https://pypi.tuna.tsinghua.edu.cn/simple`
|
||||
Run: `python -m pytest -v`
|
||||
Expected: PASS(1 passed)
|
||||
|
||||
- [ ] **Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add pyproject.toml src tests
|
||||
git commit -m "chore: 项目骨架与 pyproject 打包(src 布局 + pytest 就绪)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 2: 数据模型 data_models.py
|
||||
|
||||
**Files:**
|
||||
- Create: `src/genesis/data_models.py`
|
||||
- Test: `tests/test_data_models.py`
|
||||
|
||||
**Interfaces:**
|
||||
- Consumes: Genesis 包骨架(Task 1)
|
||||
- Produces: 模块 `genesis.data_models`,导出:
|
||||
- 枚举:`SheetType`(8) / `ElementType`(6) / `RelationType`(6) / `Confidence`(3) / `ExtractionMethod`(2)
|
||||
- 类型:`Provenance` `CellFormatting` `CellComment` `CellValue` `ExcelTable` `ChapterMarker` `ParsedTemplate` `RuleDocument` `ImageAnalysis` `ControllerInfo` `ServiceInfo` `EntityInfo` `EndpointInfo` `ExistingSystemInfo` `UnifiedDocument` `CodeStructure` `ImageDescription` `StructuredSource`(Task 3 的 config 不需要,后续里程碑消费)
|
||||
|
||||
- [ ] **Step 1: 写失败测试**
|
||||
|
||||
`tests/test_data_models.py`:
|
||||
```python
|
||||
from dataclasses import asdict
|
||||
|
||||
from genesis.data_models import (
|
||||
CellComment, CellFormatting, CellValue, Confidence, ElementType,
|
||||
ExcelTable, ExtractionMethod, ImageAnalysis, ParsedTemplate, Provenance,
|
||||
RelationType, RuleDocument, SheetType, StructuredSource,
|
||||
)
|
||||
|
||||
|
||||
def test_sheettype_has_8_members():
|
||||
assert len(SheetType) == 8
|
||||
assert SheetType.FUNCTION.value == "FUNCTION"
|
||||
assert SheetType.GENERIC.value == "GENERIC"
|
||||
|
||||
|
||||
def test_value_enum_members():
|
||||
assert ElementType.FUNCTION.value == "機能"
|
||||
assert RelationType.USE.value == "利用"
|
||||
assert Confidence.HIGH.value == "high"
|
||||
assert ExtractionMethod.OPENPYXL.value == "openpyxl"
|
||||
assert ExtractionMethod.LLM_FROM_FREE_TEXT.value == "llm_from_free_text"
|
||||
|
||||
|
||||
def test_cellformatting_defaults():
|
||||
fmt = CellFormatting()
|
||||
assert fmt.strikethrough is False
|
||||
assert fmt.font_color is None
|
||||
assert fmt.bg_color is None
|
||||
|
||||
|
||||
def test_cellvalue_forward_reference_works():
|
||||
"""CellValue 引用后置定义的 CellFormatting/CellComment(future.annotations 落地)"""
|
||||
prov = Provenance(file_name="f.xlsx", sheet_name="S", row=1, column="A", column_header="h")
|
||||
cv = CellValue(
|
||||
value="x",
|
||||
provenance=prov,
|
||||
formatting=CellFormatting(strikethrough=True),
|
||||
comment=CellComment(author="reviewer", text="check", source_uri="f.xlsx#S!A1"),
|
||||
)
|
||||
assert cv.formatting.strikethrough is True
|
||||
assert cv.comment.author == "reviewer"
|
||||
|
||||
|
||||
def test_excel_table_references_sheettype():
|
||||
table = ExcelTable(
|
||||
name="機能一覧",
|
||||
detected_type=SheetType.FUNCTION,
|
||||
extraction_method=ExtractionMethod.OPENPYXL.value,
|
||||
headers=["機能ID", "機能名"],
|
||||
rows=[],
|
||||
)
|
||||
assert table.detected_type is SheetType.FUNCTION
|
||||
assert table.extraction_method == "openpyxl"
|
||||
|
||||
|
||||
def test_structured_source_assembles_all():
|
||||
source = StructuredSource(
|
||||
tables=[],
|
||||
template=ParsedTemplate(file_name="t.docx", sections=[], placeholders={}, styles={}),
|
||||
rule_docs=[RuleDocument(
|
||||
file_name="記入規則.docx", category="write", markdown_content="# 規則",
|
||||
source_path="samples/記入規則.docx", file_type="word", hash="abc",
|
||||
)],
|
||||
image_analyses=[ImageAnalysis(
|
||||
image_ref="img1", description="画面遷移図", confidence=0.9,
|
||||
source_uri="f.xlsx#S!A1", sheet_name="S", anchor_cell="A1", status="recognized",
|
||||
)],
|
||||
existing_system=None,
|
||||
comments=[],
|
||||
)
|
||||
assert source.rule_docs[0].category == "write"
|
||||
assert source.image_analyses[0].nearby_text == ""
|
||||
|
||||
|
||||
def test_asdict_serializable():
|
||||
prov = Provenance(file_name="f.xlsx", sheet_name="S", row=1, column="A", column_header="h")
|
||||
d = asdict(CellValue(value=1, provenance=prov))
|
||||
assert d["provenance"]["row"] == 1
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 运行测试,确认失败**
|
||||
|
||||
Run: `python -m pytest tests/test_data_models.py -v`
|
||||
Expected: FAIL(`ModuleNotFoundError: No module named 'genesis.data_models'`)
|
||||
|
||||
- [ ] **Step 3: 实现 data_models.py**
|
||||
|
||||
`src/genesis/data_models.py`(完整内容):
|
||||
```python
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import Any
|
||||
|
||||
|
||||
class SheetType(Enum):
|
||||
"""Excel Sheet 的类型(Parser SheetDetector 判定结果)"""
|
||||
FUNCTION = "FUNCTION"
|
||||
SCREEN = "SCREEN"
|
||||
REPORT = "REPORT"
|
||||
DATABASE = "DATABASE"
|
||||
INTERFACE = "INTERFACE"
|
||||
BATCH = "BATCH"
|
||||
MASTER = "MASTER"
|
||||
GENERIC = "GENERIC"
|
||||
|
||||
|
||||
class ElementType(Enum):
|
||||
"""Impact Agent 抽取的构成要素类型"""
|
||||
FUNCTION = "機能"
|
||||
SCREEN = "画面"
|
||||
REPORT = "帳票"
|
||||
DB = "DB"
|
||||
IF = "IF"
|
||||
BATCH = "バッチ"
|
||||
|
||||
|
||||
class RelationType(Enum):
|
||||
"""关联类型(Impact Agent 推理结果)"""
|
||||
USE = "利用"
|
||||
REFER = "参照"
|
||||
UPDATE = "更新"
|
||||
OUTPUT = "输出"
|
||||
INPUT = "输入"
|
||||
DEPEND = "依赖"
|
||||
|
||||
|
||||
class Confidence(Enum):
|
||||
"""置信度等级"""
|
||||
HIGH = "high"
|
||||
MEDIUM = "medium"
|
||||
LOW = "low"
|
||||
|
||||
|
||||
class ExtractionMethod(Enum):
|
||||
"""Excel 表的抽取方式"""
|
||||
OPENPYXL = "openpyxl"
|
||||
LLM_FROM_FREE_TEXT = "llm_from_free_text"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Provenance:
|
||||
file_name: str
|
||||
sheet_name: str
|
||||
row: int
|
||||
column: str
|
||||
column_header: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class CellFormatting:
|
||||
strikethrough: bool = False
|
||||
font_color: str | None = None
|
||||
bg_color: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class CellComment:
|
||||
author: str
|
||||
text: str
|
||||
source_uri: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class CellValue:
|
||||
value: Any
|
||||
provenance: Provenance
|
||||
formatting: CellFormatting | None = None
|
||||
comment: CellComment | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class ExcelTable:
|
||||
name: str
|
||||
detected_type: SheetType
|
||||
extraction_method: str # 取 ExtractionMethod 的 value(同一常量来源)
|
||||
headers: list[str]
|
||||
rows: list[dict[str, CellValue]]
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChapterMarker:
|
||||
type: str # "heading" | "bookmark" | "placeholder"
|
||||
name: str
|
||||
level: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class ParsedTemplate:
|
||||
file_name: str
|
||||
sections: list[ChapterMarker]
|
||||
placeholders: dict[str, str]
|
||||
styles: dict
|
||||
|
||||
|
||||
@dataclass
|
||||
class RuleDocument:
|
||||
file_name: str
|
||||
category: str # "write" | "design" | "ref"
|
||||
markdown_content: str
|
||||
source_path: str
|
||||
file_type: str # "word" | "excel" | "ppt"
|
||||
hash: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ImageAnalysis:
|
||||
"""图片分析结果(Parser 组装,StructuredSource 消费)"""
|
||||
image_ref: str
|
||||
description: str
|
||||
confidence: float
|
||||
source_uri: str
|
||||
sheet_name: str
|
||||
anchor_cell: str
|
||||
status: str # "recognized" | "recorded_only" | "failed"
|
||||
nearby_text: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class ControllerInfo:
|
||||
name: str
|
||||
class_name: str
|
||||
path: str
|
||||
base_path: str
|
||||
endpoints: list[str]
|
||||
source_uri: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ServiceInfo:
|
||||
name: str
|
||||
class_name: str
|
||||
path: str
|
||||
methods: list[str]
|
||||
source_uri: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class EntityInfo:
|
||||
name: str
|
||||
class_name: str
|
||||
path: str
|
||||
table_name: str | None
|
||||
fields: list[str]
|
||||
source_uri: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class EndpointInfo:
|
||||
method: str
|
||||
path: str
|
||||
controller: str | None
|
||||
description: str
|
||||
source_uri: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ExistingSystemInfo:
|
||||
controller_layer: list[ControllerInfo]
|
||||
service_layer: list[ServiceInfo]
|
||||
entity_layer: list[EntityInfo]
|
||||
api_endpoints: list[EndpointInfo]
|
||||
source_path: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class UnifiedDocument:
|
||||
"""FileReader 的统一输出(多格式归一化)"""
|
||||
file_name: str
|
||||
file_type: str # "excel" | "word" | "ppt" | "text"
|
||||
source_path: str
|
||||
content_type: str
|
||||
tables: list[list[list[Any]]] | None = None
|
||||
sheet_names: list[str] | None = None
|
||||
paragraphs: list[dict] | None = None
|
||||
slides: list[dict] | None = None
|
||||
text: str | None = None
|
||||
encoding: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class CodeStructure:
|
||||
"""CodeParser 的解析输出"""
|
||||
root_path: str
|
||||
language: str
|
||||
modules: list[dict]
|
||||
classes: list[dict]
|
||||
controllers: list[ControllerInfo]
|
||||
services: list[ServiceInfo]
|
||||
entities: list[EntityInfo]
|
||||
endpoints: list[EndpointInfo]
|
||||
raw_imports: list[dict]
|
||||
|
||||
|
||||
@dataclass
|
||||
class ImageDescription:
|
||||
"""ImageAnalyzer 的原始识别输出(工具层;业务侧用 ImageAnalysis)"""
|
||||
image_ref: str
|
||||
description: str
|
||||
objects: list[str]
|
||||
ocr_text: str | None
|
||||
confidence: float
|
||||
model: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class StructuredSource:
|
||||
tables: list[ExcelTable]
|
||||
template: ParsedTemplate
|
||||
rule_docs: list[RuleDocument]
|
||||
image_analyses: list[ImageAnalysis]
|
||||
existing_system: ExistingSystemInfo | None
|
||||
comments: list[CellComment]
|
||||
```
|
||||
|
||||
(请确认文件包含 `from dataclasses import dataclass` —— 上例顶部 import 中已含,若编辑器省略请补全为 `from dataclasses import dataclass`。)
|
||||
|
||||
- [ ] **Step 4: 运行测试,确认通过**
|
||||
|
||||
Run: `python -m pytest -v`
|
||||
Expected: PASS(全部 tests,含 smoke + data_models)
|
||||
|
||||
- [ ] **Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add src/genesis/data_models.py tests/test_data_models.py
|
||||
git commit -m "feat: 数据模型 data_models(design §3+§9.4,future.annotations)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 3: 配置加载 config.py
|
||||
|
||||
**Files:**
|
||||
- Create: `src/genesis/config.py`
|
||||
- Create: `tests/fixtures/app_min.yaml`, `tests/fixtures/inference_min.yaml`, `tests/fixtures/rag_min.yaml`
|
||||
- Test: `tests/test_config.py`
|
||||
|
||||
**Interfaces:**
|
||||
- Consumes: 包骨架(Task 1)
|
||||
- Produces: 模块 `genesis.config`,导出:
|
||||
- 模型:`ServerConfig` / `AppConfig` / `InferenceConfig` / `RagConfig` / `ModelSpec` / `Settings`
|
||||
- `Settings.from_dir(config_dir: Path) -> Settings`(yaml 缺失→默认;${VAR} 展开)
|
||||
- `Settings.get_redacted() -> dict`(key/secret/token → "***")
|
||||
- 优先级:`GENESIS_` 前缀环境变量(`__` 嵌套)> yaml > 默认值
|
||||
|
||||
- [ ] **Step 1: 写 fixtures**
|
||||
|
||||
`tests/fixtures/app_min.yaml`:
|
||||
```yaml
|
||||
server:
|
||||
max_upload_mb: 10
|
||||
session:
|
||||
sqlite_path: "C:/tmp/genesis.db"
|
||||
task_queue:
|
||||
backend: memory
|
||||
timeout_sec: 300
|
||||
```
|
||||
|
||||
`tests/fixtures/inference_min.yaml`:
|
||||
```yaml
|
||||
models:
|
||||
primary:
|
||||
provider: deepseek
|
||||
name: deepseek-chat
|
||||
llm_calls:
|
||||
max_context_tokens: 16000
|
||||
structured_output:
|
||||
max_parse_retry: 3
|
||||
```
|
||||
|
||||
`tests/fixtures/rag_min.yaml`:
|
||||
```yaml
|
||||
embedding:
|
||||
model: BAAI/bge-small-zh-v1.5
|
||||
vector_store:
|
||||
adapter: chroma
|
||||
qdrant:
|
||||
url: http://qdrant:6333
|
||||
api_key: ${QDRANT_API_KEY}
|
||||
retrieval:
|
||||
rrf_k: 42
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 写失败测试**
|
||||
|
||||
`tests/test_config.py`:
|
||||
```python
|
||||
from pathlib import Path
|
||||
from genesis.config import Settings
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures"
|
||||
|
||||
|
||||
def test_from_dir_maps_yaml_fields():
|
||||
s = Settings.from_dir(FIXTURES)
|
||||
assert s.app.server.max_upload_mb == 100
|
||||
assert s.app.session["sqlite_path"] == "C:/tmp/genesis.db"
|
||||
assert s.app.task_queue["timeout_sec"] == 300
|
||||
assert s.inference.models.primary.name == "deepseek-chat"
|
||||
assert s.inference.llm_calls.max_context_tokens == 16000
|
||||
assert s.inference.structured_output.max_parse_retry == 3
|
||||
assert s.rag.embedding.model == "BAAI/bge-small-zh-v1.5"
|
||||
assert s.rag.retrieval.rrf_k == 42
|
||||
|
||||
|
||||
def test_defaults_when_dir_empty(tmp_path):
|
||||
s = Settings.from_dir(tmp_path)
|
||||
assert s.app.name == "genesis"
|
||||
assert s.app.server.max_upload_mb == 100
|
||||
assert s.app.task_queue["backend"] == "memory"
|
||||
assert s.inference.models.primary.name == "deepseek-chat"
|
||||
assert s.rag.embedding.model == "BAAI/bge-small-zh-v1.5"
|
||||
assert s.rag.retrieval.rrf_k == 60
|
||||
|
||||
|
||||
def test_env_override_yaml(monkeypatch):
|
||||
monkeypatch.setenv("GENESIS_APP__SERVER__MAX_UPLOAD_MB", "25")
|
||||
s = Settings.from_dir(FIXTURES)
|
||||
assert s.app.server.max_upload_mb == 25
|
||||
|
||||
|
||||
def test_env_nested_creation(monkeypatch):
|
||||
monkeypatch.setenv("GENESIS_RAG__RETRIEVAL__DEFAULT_TOP_K", "7")
|
||||
s = Settings.from_dir(FIXTURES)
|
||||
assert s.rag.retrieval.default_top_k == 7
|
||||
|
||||
|
||||
def test_env_placeholder_expansion(monkeypatch):
|
||||
monkeypatch.setenv("QDRANT_API_KEY", "sk-test-xyz")
|
||||
s = Settings.from_dir(FIXTURES)
|
||||
assert s.rag.vector_store.qdrant.api_key == "sk-test-xyz"
|
||||
|
||||
|
||||
def test_redacted_hides_secrets():
|
||||
s = Settings.from_dir(FIXTURES)
|
||||
red = s.get_redacted()
|
||||
assert "sk-test-xyz" not in str(red)
|
||||
assert red["rag"]["vector_store"]["qdrant"]["api_key"] == "***"
|
||||
```
|
||||
|
||||
- [ ] **Step 3: 运行测试,确认失败**
|
||||
|
||||
Run: `python -m pytest tests/test_config.py -v`
|
||||
Expected: FAIL(`ModuleNotFoundError: No module named 'genesis.config'`)
|
||||
|
||||
- [ ] **Step 4: 实现 config.py**
|
||||
|
||||
`src/genesis/config.py`(完整内容):
|
||||
```python
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
from pydantic import BaseModel, Field
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
|
||||
SECRET_KEYWORDS = ("key", "secret", "token")
|
||||
ENV_PREFIX = "GENESIS_"
|
||||
|
||||
|
||||
# ---------- 各 yaml 对应的 pydantic 模型 ----------
|
||||
|
||||
class ServerConfig(BaseModel):
|
||||
max_upload_mb: int = 100
|
||||
allowed_extensions: list[str] = Field(
|
||||
default_factory=lambda: [".xlsx", ".xls", ".docx", ".pptx", ".java", ".xml", ".yml"]
|
||||
)
|
||||
|
||||
|
||||
class AppConfig(BaseModel):
|
||||
name: str = "genesis"
|
||||
version: str = "0.1.0"
|
||||
timezone: str = "Asia/Tokyo"
|
||||
server: ServerConfig = Field(default_factory=ServerConfig)
|
||||
session: dict[str, Any] = Field(default_factory=lambda: {
|
||||
"sqlite_path": "/data/db/genesis.db",
|
||||
"snapshot_dir": "/data/db/snapshots",
|
||||
})
|
||||
paths: dict[str, Any] = Field(default_factory=lambda: {
|
||||
"user_root": "/data/users",
|
||||
"shared_root": "/data/shared",
|
||||
})
|
||||
task_queue: dict[str, Any] = Field(default_factory=lambda: {
|
||||
"backend": "memory",
|
||||
"redis_url": "",
|
||||
"timeout_sec": 600,
|
||||
"retry_default": 2,
|
||||
})
|
||||
|
||||
|
||||
class ModelSpec(BaseModel):
|
||||
provider: str = "deepseek"
|
||||
name: str = "deepseek-chat"
|
||||
temperature: float = 0.2
|
||||
max_tokens: int = 4096
|
||||
timeout_sec: int = 60
|
||||
retry_backoff: list[float] = Field(default_factory=lambda: [1.0, 3.0, 7.0])
|
||||
|
||||
|
||||
class InferenceModels(BaseModel):
|
||||
primary: ModelSpec = Field(default_factory=ModelSpec)
|
||||
fallback: ModelSpec = Field(default_factory=lambda: ModelSpec(provider="qwen", name="qwen-max"))
|
||||
vision: ModelSpec = Field(default_factory=lambda: ModelSpec(name="deepseek-vl", timeout_sec=90))
|
||||
|
||||
|
||||
class LlmCallsConfig(BaseModel):
|
||||
token_estimation: str = "tiktoken"
|
||||
max_context_tokens: int = 32000
|
||||
truncation_policy: dict[str, Any] = Field(default_factory=lambda: {
|
||||
"priority": ["shrink_rule_chunks", "summarize_history", "truncate_data"],
|
||||
})
|
||||
|
||||
|
||||
class StructuredOutputConfig(BaseModel):
|
||||
max_parse_retry: int = 2
|
||||
|
||||
|
||||
class PromptRegistryConfig(BaseModel):
|
||||
prompts_dir: str = "./prompts"
|
||||
default_version: str = "latest"
|
||||
|
||||
|
||||
class InferenceConfig(BaseModel):
|
||||
models: InferenceModels = Field(default_factory=InferenceModels)
|
||||
llm_calls: LlmCallsConfig = Field(default_factory=LlmCallsConfig)
|
||||
structured_output: StructuredOutputConfig = Field(default_factory=StructuredOutputConfig)
|
||||
prompt_registry: PromptRegistryConfig = Field(default_factory=PromptRegistryConfig)
|
||||
|
||||
|
||||
class EmbeddingConfig(BaseModel):
|
||||
model: str = "BAAI/bge-small-zh-v1.5"
|
||||
device: str = "cpu"
|
||||
max_batch_size: int = 32
|
||||
cache_dir: str = "/data/shared/models"
|
||||
|
||||
|
||||
class ChromaStoreConfig(BaseModel):
|
||||
persist_dir: str = "/data/shared/rules-handbook/chroma"
|
||||
|
||||
|
||||
class QdrantStoreConfig(BaseModel):
|
||||
url: str = "http://qdrant:6333"
|
||||
api_key: str = ""
|
||||
|
||||
|
||||
class VectorStoreConfig(BaseModel):
|
||||
adapter: str = "chroma"
|
||||
chroma: ChromaStoreConfig = Field(default_factory=ChromaStoreConfig)
|
||||
qdrant: QdrantStoreConfig = Field(default_factory=QdrantStoreConfig)
|
||||
|
||||
|
||||
class ChunkingConfig(BaseModel):
|
||||
word_max_tokens: int = 512
|
||||
excel_rule_block_rows: int = 10
|
||||
ppt_pages_per_chunk: int = 2
|
||||
min_tokens: int = 30
|
||||
|
||||
|
||||
class RetrievalConfig(BaseModel):
|
||||
channel_top_k: int = 10
|
||||
rrf_k: int = 60
|
||||
default_top_k: int = 5
|
||||
contextual_enrichment: bool = True
|
||||
|
||||
|
||||
class RagConfig(BaseModel):
|
||||
embedding: EmbeddingConfig = Field(default_factory=EmbeddingConfig)
|
||||
vector_store: VectorStoreConfig = Field(default_factory=VectorStoreConfig)
|
||||
chunking: ChunkingConfig = Field(default_factory=ChunkingConfig)
|
||||
retrieval: RetrievalConfig = Field(default_factory=RetrievalConfig)
|
||||
|
||||
|
||||
# ---------- 加载辅助 ----------
|
||||
|
||||
def _expand_env(data: Any) -> Any:
|
||||
"""递归展开 ${VAR} 占位(读环境变量,缺失→空串)"""
|
||||
if isinstance(data, dict):
|
||||
return {k: _expand_env(v) for k, v in data.items()}
|
||||
if isinstance(data, list):
|
||||
return [_expand_env(v) for v in data]
|
||||
if isinstance(data, str) and data.startswith("${") and data.endswith("}"):
|
||||
return os.environ.get(data[2:-1], "")
|
||||
return data
|
||||
|
||||
|
||||
def _deep_merge(base: dict, override: dict) -> dict:
|
||||
"""递归合并:override 覆盖 base;非 dict 值直接取 override 存在者"""
|
||||
out = dict(base)
|
||||
for k, v in override.items():
|
||||
if isinstance(v, dict) and isinstance(out.get(k), dict):
|
||||
out[k] = _deep_merge(out[k], v)
|
||||
else:
|
||||
out[k] = v
|
||||
return out
|
||||
|
||||
|
||||
def _env_overrides() -> dict:
|
||||
"""收集 GENESIS_ 前缀的条目为嵌套 dict,__ 为嵌套分隔"""
|
||||
result: dict[str, Any] = {}
|
||||
for key, value in os.environ.items():
|
||||
if key.startswith(ENV_PREFIX):
|
||||
parts = key[len(ENV_PREFIX):].split("__")
|
||||
node = result
|
||||
for part in parts[:-1]:
|
||||
node = node.setdefault(part, {})
|
||||
node[parts[-1]] = value
|
||||
return result
|
||||
|
||||
|
||||
def _load_yaml(config_dir: Path, name: str) -> dict:
|
||||
path = config_dir / f"{name}.yaml"
|
||||
if not path.exists():
|
||||
return {}
|
||||
with path.open("r", encoding="utf-8") as f:
|
||||
return yaml.safe_load(f) or {}
|
||||
|
||||
|
||||
def _redact(data: dict) -> dict:
|
||||
out = {}
|
||||
for k, v in data.items():
|
||||
if any(kw in str(k).lower() for kw in SECRET_KEYWORDS):
|
||||
out[k] = "***"
|
||||
elif isinstance(v, dict):
|
||||
out[k] = _redact(v)
|
||||
else:
|
||||
out[k] = v
|
||||
return out
|
||||
|
||||
|
||||
# ---------- 根 Settings ----------
|
||||
|
||||
class Settings(BaseSettings):
|
||||
model_config = SettingsConfigDict(env_prefix=ENV_PREFIX, env_file=".env", extra="ignore")
|
||||
|
||||
app: AppConfig = Field(default_factory=AppConfig)
|
||||
inference: InferenceConfig = Field(default_factory=InferenceConfig)
|
||||
rag: RagConfig = Field(default_factory=RagConfig)
|
||||
|
||||
@classmethod
|
||||
def from_dir(cls, config_dir: Path | str) -> "Settings":
|
||||
config_dir = Path(config_dir)
|
||||
raw = {
|
||||
"app": _load_yaml(config_dir, "app"),
|
||||
"inference": _load_yaml(config_dir, "inference"),
|
||||
"rag": _load_yaml(config_dir, "rag"),
|
||||
}
|
||||
env = _env_overrides()
|
||||
merged = {k: _deep_merge(raw[k], env.get(k, {})) for k in raw}
|
||||
return cls(**{k: _expand_env(v) for k, v in merged.items()})
|
||||
|
||||
def get_redacted(self) -> dict:
|
||||
return _redact(self.model_dump(mode="json"))
|
||||
```
|
||||
|
||||
> 顺序说明:本文件类定义已是自顶向下的依赖顺序(`ModelSpec` → `InferenceModels` → `InferenceConfig`;`RetrievalConfig` → `RagConfig`),**不要重排**。
|
||||
|
||||
- [ ] **Step 5: 运行测试,确认通过**
|
||||
|
||||
Run: `python -m pytest -v`
|
||||
Expected: PASS(全部含 test_config)
|
||||
|
||||
- [ ] **Step 6: 修复可能出现的 NameError**
|
||||
|
||||
若 `python -m pytest` 报 `NameError: name 'X' is not defined`,说明类顺序不符;按 Step 4 最后的顺序提示调整(前置类型先行),重跑直至全绿。
|
||||
|
||||
- [ ] **Step 7: Commit**
|
||||
|
||||
```bash
|
||||
git add src/genesis/config.py tests/test_config.py tests/fixtures
|
||||
git commit -m "feat: 配置加载 config(三 yaml + env 优先级 + 脱敏)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Self-Review 结论(写计划时已执行)
|
||||
|
||||
- **Spec 覆盖**:设计文档 §2(结构)→ Task1;§3(数据模型)→ Task2;§4(config)→ Task3;§5(测试策略)→ 内嵌于各 Task;§6(完成门槛)→ 最终 pytest 全绿即满足。
|
||||
- **占位符扫描**:无 TBD/TODO;每个 Step 有完整代码与命令。
|
||||
- **类型一致性**:`Settings.from_dir` / `get_redacted` / 各模型名在 Task3 内统一;Task2 导出类型与 design 一致;Task3 测试中 `s.app.task_queue` 为 dict(与 yaml 结构一致)、`s.app.session` 为 dict、`s.rag.vector_store.qdrant.api_key` 为 str。
|
||||
Reference in New Issue
Block a user