chore: 문서를 작성할 때 한국어의 표현 작성 스킬 추가 및 1인칭 관점의 글 작성 검증 테스트 추가
This commit is contained in:
DongHyeonka
@@ -0,0 +1,676 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import re
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable
|
||||
|
||||
|
||||
class ValidationError(ValueError):
|
||||
"""Raised when a user-supplied contract is invalid."""
|
||||
|
||||
|
||||
class DocumentType(str, Enum):
|
||||
TECHNICAL_BLOG = "technical_blog"
|
||||
TUTORIAL = "tutorial"
|
||||
HOW_TO = "how_to"
|
||||
EXPLANATION = "explanation"
|
||||
REFERENCE = "reference"
|
||||
TROUBLESHOOTING = "troubleshooting"
|
||||
DESIGN_DOC = "design_doc"
|
||||
|
||||
@classmethod
|
||||
def values(cls) -> list[str]:
|
||||
return [member.value for member in cls]
|
||||
|
||||
|
||||
class Severity(str, Enum):
|
||||
BLOCKER = "blocker"
|
||||
ERROR = "error"
|
||||
WARNING = "warning"
|
||||
INFO = "info"
|
||||
|
||||
|
||||
REVIEW_DIMENSIONS: tuple[str, ...] = (
|
||||
"reader_goal_alignment",
|
||||
"information_architecture",
|
||||
"logical_flow",
|
||||
"decision_rationale",
|
||||
"source_usefulness",
|
||||
"reader_facing_prose",
|
||||
"cognitive_load",
|
||||
"evidence_traceability",
|
||||
"example_verifiability",
|
||||
"scannability",
|
||||
"operational_safety",
|
||||
"completeness_and_limits",
|
||||
)
|
||||
|
||||
REVIEW_SEVERITIES = frozenset(member.value for member in Severity)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Audience:
|
||||
roles: list[str]
|
||||
prior_knowledge: list[str] = field(default_factory=list)
|
||||
needs: list[str] = field(default_factory=list)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any]) -> "Audience":
|
||||
roles = _string_list(data.get("roles"), "audience.roles", required=True)
|
||||
return cls(
|
||||
roles=roles,
|
||||
prior_knowledge=_string_list(data.get("prior_knowledge", []), "audience.prior_knowledge"),
|
||||
needs=_string_list(data.get("needs", []), "audience.needs"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Constraints:
|
||||
target_words: int = 1600
|
||||
tone: str = "professional and direct"
|
||||
version_context: str = ""
|
||||
max_heading_depth: int = 3
|
||||
require_citations: bool = True
|
||||
allow_external_knowledge: bool = False
|
||||
citation_style: str = "hidden"
|
||||
date_policy: str = "only_when_material"
|
||||
style_profile: str = "auto"
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any] | None) -> "Constraints":
|
||||
if data is None:
|
||||
data = {}
|
||||
if not isinstance(data, dict):
|
||||
raise ValidationError("constraints must be an object")
|
||||
target_words = _integer(data.get("target_words", 1600), "constraints.target_words")
|
||||
max_heading_depth = _integer(data.get("max_heading_depth", 3), "constraints.max_heading_depth")
|
||||
if target_words < 200 or target_words > 30000:
|
||||
raise ValidationError("constraints.target_words must be between 200 and 30000")
|
||||
if max_heading_depth < 2 or max_heading_depth > 6:
|
||||
raise ValidationError("constraints.max_heading_depth must be between 2 and 6")
|
||||
citation_style = str(data.get("citation_style", "hidden")).strip().lower()
|
||||
if citation_style not in {"hidden", "footnote", "inline_link", "source_id"}:
|
||||
raise ValidationError(
|
||||
"constraints.citation_style must be one of: hidden, footnote, inline_link, source_id"
|
||||
)
|
||||
date_policy = str(data.get("date_policy", "only_when_material")).strip().lower()
|
||||
if date_policy not in {"only_when_material", "always", "never"}:
|
||||
raise ValidationError(
|
||||
"constraints.date_policy must be one of: only_when_material, always, never"
|
||||
)
|
||||
return cls(
|
||||
target_words=target_words,
|
||||
tone=_nonempty_string(data.get("tone", "professional and direct"), "constraints.tone"),
|
||||
version_context=str(data.get("version_context", "")).strip(),
|
||||
max_heading_depth=max_heading_depth,
|
||||
require_citations=_boolean(data.get("require_citations", True), "constraints.require_citations"),
|
||||
allow_external_knowledge=_boolean(
|
||||
data.get("allow_external_knowledge", False),
|
||||
"constraints.allow_external_knowledge",
|
||||
),
|
||||
citation_style=citation_style,
|
||||
date_policy=date_policy,
|
||||
style_profile=str(data.get("style_profile", "auto")).strip() or "auto",
|
||||
)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Brief:
|
||||
title: str
|
||||
document_type: DocumentType
|
||||
language: str
|
||||
audience: Audience
|
||||
reader_goal: str
|
||||
core_message: str
|
||||
scope: list[str]
|
||||
non_scope: list[str]
|
||||
prerequisites: list[str]
|
||||
required_topics: list[str]
|
||||
constraints: Constraints = field(default_factory=Constraints)
|
||||
forbidden_claims: list[str] = field(default_factory=list)
|
||||
metadata: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any]) -> "Brief":
|
||||
if not isinstance(data, dict):
|
||||
raise ValidationError("brief must be a JSON object")
|
||||
raw_type = _nonempty_string(data.get("document_type"), "document_type")
|
||||
try:
|
||||
document_type = DocumentType(raw_type)
|
||||
except ValueError as exc:
|
||||
raise ValidationError(
|
||||
f"document_type must be one of: {', '.join(DocumentType.values())}"
|
||||
) from exc
|
||||
return cls(
|
||||
title=_nonempty_string(data.get("title"), "title"),
|
||||
document_type=document_type,
|
||||
language=_nonempty_string(data.get("language", "ko-KR"), "language"),
|
||||
audience=Audience.from_dict(_mapping(data.get("audience"), "audience")),
|
||||
reader_goal=_nonempty_string(data.get("reader_goal"), "reader_goal"),
|
||||
core_message=_nonempty_string(data.get("core_message"), "core_message"),
|
||||
scope=_string_list(data.get("scope"), "scope", required=True),
|
||||
non_scope=_string_list(data.get("non_scope", []), "non_scope"),
|
||||
prerequisites=_string_list(data.get("prerequisites", []), "prerequisites"),
|
||||
required_topics=_string_list(data.get("required_topics", []), "required_topics"),
|
||||
constraints=Constraints.from_dict(data.get("constraints")),
|
||||
forbidden_claims=_string_list(data.get("forbidden_claims", []), "forbidden_claims"),
|
||||
metadata=_mapping(data.get("metadata", {}), "metadata"),
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
data = asdict(self)
|
||||
data["document_type"] = self.document_type.value
|
||||
return data
|
||||
|
||||
@property
|
||||
def is_korean(self) -> bool:
|
||||
return self.language.lower().startswith("ko")
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Source:
|
||||
id: str
|
||||
title: str
|
||||
url: str
|
||||
publisher: str = ""
|
||||
accessed: str = ""
|
||||
facts: list[str] = field(default_factory=list)
|
||||
notes: str = ""
|
||||
source_type: str = "external"
|
||||
status: str = ""
|
||||
path: str = ""
|
||||
heading: str = ""
|
||||
line_start: int | None = None
|
||||
line_end: int | None = None
|
||||
claim_ids: list[str] = field(default_factory=list)
|
||||
decision_ids: list[str] = field(default_factory=list)
|
||||
priority: float = 0.0
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any]) -> "Source":
|
||||
source_id = _nonempty_string(data.get("id"), "source.id")
|
||||
if not re.fullmatch(r"[A-Za-z0-9_-]+", source_id):
|
||||
raise ValidationError(f"source id contains unsupported characters: {source_id}")
|
||||
line_start = _optional_integer(data.get("line_start"), f"source[{source_id}].line_start")
|
||||
line_end = _optional_integer(data.get("line_end"), f"source[{source_id}].line_end")
|
||||
if line_start is not None and line_start < 1:
|
||||
raise ValidationError(f"source[{source_id}].line_start must be positive")
|
||||
if line_end is not None and line_end < 1:
|
||||
raise ValidationError(f"source[{source_id}].line_end must be positive")
|
||||
if line_start is not None and line_end is not None and line_end < line_start:
|
||||
raise ValidationError(f"source[{source_id}].line_end must be >= line_start")
|
||||
return cls(
|
||||
id=source_id,
|
||||
title=_nonempty_string(data.get("title"), f"source[{source_id}].title"),
|
||||
url=_nonempty_string(data.get("url"), f"source[{source_id}].url"),
|
||||
publisher=str(data.get("publisher", "")).strip(),
|
||||
accessed=str(data.get("accessed", "")).strip(),
|
||||
facts=_string_list(data.get("facts", []), f"source[{source_id}].facts"),
|
||||
notes=str(data.get("notes", "")).strip(),
|
||||
source_type=str(data.get("source_type", "external")).strip() or "external",
|
||||
status=str(data.get("status", "")).strip(),
|
||||
path=str(data.get("path", "")).strip(),
|
||||
heading=str(data.get("heading", "")).strip(),
|
||||
line_start=line_start,
|
||||
line_end=line_end,
|
||||
claim_ids=_string_list(data.get("claim_ids", []), f"source[{source_id}].claim_ids"),
|
||||
decision_ids=_string_list(data.get("decision_ids", []), f"source[{source_id}].decision_ids"),
|
||||
priority=_number(data.get("priority", 0.0), f"source[{source_id}].priority"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class SourcePack:
|
||||
sources: list[Source] = field(default_factory=list)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any] | None) -> "SourcePack":
|
||||
if data is None:
|
||||
data = {"sources": []}
|
||||
if not isinstance(data, dict):
|
||||
raise ValidationError("source pack must be a JSON object")
|
||||
raw_sources = data.get("sources", [])
|
||||
if not isinstance(raw_sources, list):
|
||||
raise ValidationError("sources must be an array")
|
||||
sources = [Source.from_dict(_mapping(item, "source")) for item in raw_sources]
|
||||
ids = [source.id for source in sources]
|
||||
duplicates = sorted({source_id for source_id in ids if ids.count(source_id) > 1})
|
||||
if duplicates:
|
||||
raise ValidationError(f"duplicate source ids: {', '.join(duplicates)}")
|
||||
return cls(sources=sources)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {"sources": [asdict(source) for source in self.sources]}
|
||||
|
||||
@property
|
||||
def ids(self) -> set[str]:
|
||||
return {source.id for source in self.sources}
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class OutlineSection:
|
||||
id: str
|
||||
intent: str
|
||||
title: str
|
||||
reader_question: str
|
||||
purpose: str
|
||||
must_include: list[str] = field(default_factory=list)
|
||||
evidence_ids: list[str] = field(default_factory=list)
|
||||
decision_requirements: list[str] = field(default_factory=list)
|
||||
transition_to_next: str = ""
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any]) -> "OutlineSection":
|
||||
return cls(
|
||||
id=_nonempty_string(data.get("id"), "outline.section.id"),
|
||||
intent=_nonempty_string(data.get("intent"), "outline.section.intent"),
|
||||
title=_nonempty_string(data.get("title"), "outline.section.title"),
|
||||
reader_question=_nonempty_string(data.get("reader_question"), "outline.section.reader_question"),
|
||||
purpose=_nonempty_string(data.get("purpose"), "outline.section.purpose"),
|
||||
must_include=_string_list(data.get("must_include", []), "outline.section.must_include"),
|
||||
evidence_ids=_string_list(data.get("evidence_ids", []), "outline.section.evidence_ids"),
|
||||
decision_requirements=_string_list(
|
||||
data.get("decision_requirements", []), "outline.section.decision_requirements"
|
||||
),
|
||||
transition_to_next=str(data.get("transition_to_next", "")).strip(),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Outline:
|
||||
title: str
|
||||
document_type: DocumentType
|
||||
sections: list[OutlineSection]
|
||||
planning_notes: list[str] = field(default_factory=list)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any]) -> "Outline":
|
||||
raw_type = _nonempty_string(data.get("document_type"), "outline.document_type")
|
||||
try:
|
||||
document_type = DocumentType(raw_type)
|
||||
except ValueError as exc:
|
||||
raise ValidationError(f"invalid outline document_type: {raw_type}") from exc
|
||||
raw_sections = data.get("sections")
|
||||
if not isinstance(raw_sections, list) or not raw_sections:
|
||||
raise ValidationError("outline.sections must be a non-empty array")
|
||||
sections = [OutlineSection.from_dict(_mapping(item, "outline.section")) for item in raw_sections]
|
||||
return cls(
|
||||
title=_nonempty_string(data.get("title"), "outline.title"),
|
||||
document_type=document_type,
|
||||
sections=sections,
|
||||
planning_notes=_string_list(data.get("planning_notes", []), "outline.planning_notes"),
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"title": self.title,
|
||||
"document_type": self.document_type.value,
|
||||
"sections": [asdict(section) for section in self.sections],
|
||||
"planning_notes": self.planning_notes,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class LintIssue:
|
||||
code: str
|
||||
severity: Severity
|
||||
message: str
|
||||
line: int | None = None
|
||||
section: str = ""
|
||||
suggestion: str = ""
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
data = asdict(self)
|
||||
data["severity"] = self.severity.value
|
||||
return data
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class LintReport:
|
||||
score: float
|
||||
word_count: int
|
||||
issues: list[LintIssue]
|
||||
metrics: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"score": self.score,
|
||||
"word_count": self.word_count,
|
||||
"issues": [issue.to_dict() for issue in self.issues],
|
||||
"metrics": self.metrics,
|
||||
}
|
||||
|
||||
def count(self, severity: Severity) -> int:
|
||||
return sum(issue.severity == severity for issue in self.issues)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ReviewIssue:
|
||||
section: str
|
||||
problem: str
|
||||
why_it_matters: str
|
||||
fix: str
|
||||
severity: str = "error"
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any]) -> "ReviewIssue":
|
||||
severity = _nonempty_string(data.get("severity"), "review.issue.severity").lower()
|
||||
if severity not in REVIEW_SEVERITIES:
|
||||
raise ValidationError(
|
||||
"review.issue.severity must be one of: " + ", ".join(sorted(REVIEW_SEVERITIES))
|
||||
)
|
||||
return cls(
|
||||
section=str(data.get("section", "")).strip(),
|
||||
problem=_nonempty_string(data.get("problem"), "review.issue.problem"),
|
||||
why_it_matters=_nonempty_string(
|
||||
data.get("why_it_matters"), "review.issue.why_it_matters"
|
||||
),
|
||||
fix=_nonempty_string(data.get("fix"), "review.issue.fix"),
|
||||
severity=severity,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ModelReview:
|
||||
role: str
|
||||
provider: str
|
||||
score: float
|
||||
dimension_scores: dict[str, float]
|
||||
issues: list[ReviewIssue]
|
||||
strengths: list[str]
|
||||
questions: list[str]
|
||||
raw_response: str = ""
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any], *, role: str, provider: str, raw_response: str = "") -> "ModelReview":
|
||||
if not isinstance(data, dict):
|
||||
raise ValidationError("review must be a JSON object")
|
||||
expected_top_level = {"score", "dimension_scores", "issues", "strengths", "questions"}
|
||||
missing = sorted(expected_top_level - set(data))
|
||||
unknown = sorted(set(data) - expected_top_level)
|
||||
if missing:
|
||||
raise ValidationError(f"review is missing required fields: {', '.join(missing)}")
|
||||
if unknown:
|
||||
raise ValidationError(f"review contains unsupported fields: {', '.join(unknown)}")
|
||||
|
||||
score = _number(data.get("score"), "review.score")
|
||||
if score < 0 or score > 100:
|
||||
raise ValidationError("review.score must be between 0 and 100")
|
||||
raw_dimensions = _mapping(data.get("dimension_scores"), "review.dimension_scores")
|
||||
missing_dimensions = sorted(set(REVIEW_DIMENSIONS) - set(raw_dimensions))
|
||||
unknown_dimensions = sorted(set(raw_dimensions) - set(REVIEW_DIMENSIONS))
|
||||
if missing_dimensions:
|
||||
raise ValidationError(
|
||||
"review.dimension_scores is missing: " + ", ".join(missing_dimensions)
|
||||
)
|
||||
if unknown_dimensions:
|
||||
raise ValidationError(
|
||||
"review.dimension_scores contains unsupported dimensions: "
|
||||
+ ", ".join(unknown_dimensions)
|
||||
)
|
||||
dimensions: dict[str, float] = {}
|
||||
for key in REVIEW_DIMENSIONS:
|
||||
numeric = _number(raw_dimensions[key], f"review.dimension_scores.{key}")
|
||||
if numeric < 0 or numeric > 100:
|
||||
raise ValidationError(f"review dimension {key} must be between 0 and 100")
|
||||
dimensions[key] = numeric
|
||||
raw_issues = data.get("issues")
|
||||
if not isinstance(raw_issues, list):
|
||||
raise ValidationError("review.issues must be an array")
|
||||
return cls(
|
||||
role=role,
|
||||
provider=provider,
|
||||
score=score,
|
||||
dimension_scores=dimensions,
|
||||
issues=[ReviewIssue.from_dict(_mapping(item, "review.issue")) for item in raw_issues],
|
||||
strengths=_string_list(data.get("strengths", []), "review.strengths"),
|
||||
questions=_string_list(data.get("questions", []), "review.questions"),
|
||||
raw_response=raw_response,
|
||||
)
|
||||
|
||||
@property
|
||||
def blocker_count(self) -> int:
|
||||
return sum(issue.severity == "blocker" for issue in self.issues)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"role": self.role,
|
||||
"provider": self.provider,
|
||||
"score": self.score,
|
||||
"dimension_scores": self.dimension_scores,
|
||||
"issues": [asdict(issue) for issue in self.issues],
|
||||
"strengths": self.strengths,
|
||||
"questions": self.questions,
|
||||
"raw_response": self.raw_response,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ProviderSpec:
|
||||
provider: str
|
||||
model: str = ""
|
||||
timeout_seconds: int = 300
|
||||
options: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any] | str | None, *, default: str = "mock") -> "ProviderSpec":
|
||||
if data is None:
|
||||
return cls(provider=default)
|
||||
if isinstance(data, str):
|
||||
return cls(provider=data)
|
||||
if not isinstance(data, dict):
|
||||
raise ValidationError("provider configuration must be a string or object")
|
||||
timeout = _integer(data.get("timeout_seconds", 300), "provider.timeout_seconds")
|
||||
if timeout < 1:
|
||||
raise ValidationError("provider timeout_seconds must be positive")
|
||||
return cls(
|
||||
provider=_nonempty_string(data.get("provider", default), "provider.provider"),
|
||||
model=str(data.get("model", "")).strip(),
|
||||
timeout_seconds=timeout,
|
||||
options=_mapping(data.get("options", {}), "provider.options"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ReviewerSpec:
|
||||
role: str
|
||||
provider: ProviderSpec
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any]) -> "ReviewerSpec":
|
||||
return cls(
|
||||
role=_nonempty_string(data.get("role"), "reviewer.role"),
|
||||
provider=ProviderSpec.from_dict(data),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class QualityGate:
|
||||
minimum_score: float = 82.0
|
||||
max_blockers: int = 0
|
||||
max_errors: int = 2
|
||||
max_revisions: int = 2
|
||||
deterministic_weight: float = 0.4
|
||||
model_weight: float = 0.6
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any] | None) -> "QualityGate":
|
||||
if data is None:
|
||||
data = {}
|
||||
if not isinstance(data, dict):
|
||||
raise ValidationError("quality_gate must be an object")
|
||||
minimum_score = _number(data.get("minimum_score", 82.0), "quality_gate.minimum_score")
|
||||
max_blockers = _integer(data.get("max_blockers", 0), "quality_gate.max_blockers")
|
||||
max_errors = _integer(data.get("max_errors", 2), "quality_gate.max_errors")
|
||||
max_revisions = _integer(data.get("max_revisions", 2), "quality_gate.max_revisions")
|
||||
deterministic_weight = _number(
|
||||
data.get("deterministic_weight", 0.4), "quality_gate.deterministic_weight"
|
||||
)
|
||||
model_weight = _number(data.get("model_weight", 0.6), "quality_gate.model_weight")
|
||||
if minimum_score < 0 or minimum_score > 100:
|
||||
raise ValidationError("quality_gate.minimum_score must be between 0 and 100")
|
||||
if min(max_blockers, max_errors, max_revisions) < 0:
|
||||
raise ValidationError("quality_gate count limits must be non-negative")
|
||||
if not 0 <= deterministic_weight <= 1 or not 0 <= model_weight <= 1:
|
||||
raise ValidationError("quality_gate weights must be between 0 and 1")
|
||||
if abs((deterministic_weight + model_weight) - 1.0) > 1e-6:
|
||||
raise ValidationError("quality_gate weights must sum to 1.0")
|
||||
return cls(
|
||||
minimum_score=minimum_score,
|
||||
max_blockers=max_blockers,
|
||||
max_errors=max_errors,
|
||||
max_revisions=max_revisions,
|
||||
deterministic_weight=deterministic_weight,
|
||||
model_weight=model_weight,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class PipelineConfig:
|
||||
planner: ProviderSpec
|
||||
writer: ProviderSpec
|
||||
reviewers: list[ReviewerSpec]
|
||||
reviser: ProviderSpec
|
||||
quality_gate: QualityGate
|
||||
fail_on_reviewer_error: bool = True
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any]) -> "PipelineConfig":
|
||||
if not isinstance(data, dict):
|
||||
raise ValidationError("pipeline configuration must be a JSON object")
|
||||
missing_stages = [name for name in ("planner", "writer", "reviewers", "reviser") if name not in data]
|
||||
if missing_stages:
|
||||
raise ValidationError(
|
||||
"pipeline configuration is missing required fields: " + ", ".join(missing_stages)
|
||||
)
|
||||
raw_reviewers = data.get("reviewers")
|
||||
if not isinstance(raw_reviewers, list):
|
||||
raise ValidationError("reviewers must be an array")
|
||||
reviewers = [ReviewerSpec.from_dict(_mapping(item, "reviewer")) for item in raw_reviewers]
|
||||
if not reviewers:
|
||||
raise ValidationError("reviewers must contain at least one reviewer")
|
||||
roles = [reviewer.role for reviewer in reviewers]
|
||||
duplicate_roles = sorted({role for role in roles if roles.count(role) > 1})
|
||||
if duplicate_roles:
|
||||
raise ValidationError("duplicate reviewer roles: " + ", ".join(duplicate_roles))
|
||||
return cls(
|
||||
planner=ProviderSpec.from_dict(data.get("planner")),
|
||||
writer=ProviderSpec.from_dict(data.get("writer")),
|
||||
reviewers=reviewers,
|
||||
reviser=ProviderSpec.from_dict(data.get("reviser")),
|
||||
quality_gate=QualityGate.from_dict(data.get("quality_gate")),
|
||||
fail_on_reviewer_error=_boolean(
|
||||
data.get("fail_on_reviewer_error", True), "fail_on_reviewer_error"
|
||||
),
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"planner": asdict(self.planner),
|
||||
"writer": asdict(self.writer),
|
||||
"reviewers": [
|
||||
{"role": reviewer.role, **asdict(reviewer.provider)} for reviewer in self.reviewers
|
||||
],
|
||||
"reviser": asdict(self.reviser),
|
||||
"quality_gate": asdict(self.quality_gate),
|
||||
"fail_on_reviewer_error": self.fail_on_reviewer_error,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class RoundResult:
|
||||
round_number: int
|
||||
draft_path: Path
|
||||
lint_report: LintReport
|
||||
reviews: list[ModelReview]
|
||||
composite_score: float
|
||||
blocker_count: int
|
||||
error_count: int
|
||||
passed: bool
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class RunResult:
|
||||
output_dir: Path
|
||||
final_path: Path
|
||||
report_path: Path
|
||||
manifest_path: Path
|
||||
passed: bool
|
||||
final_score: float
|
||||
rounds: list[RoundResult]
|
||||
warnings: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
def _nonempty_string(value: Any, field_name: str) -> str:
|
||||
if value is None:
|
||||
raise ValidationError(f"{field_name} is required")
|
||||
text = str(value).strip()
|
||||
if not text:
|
||||
raise ValidationError(f"{field_name} must not be empty")
|
||||
return text
|
||||
|
||||
|
||||
def _string_list(value: Any, field_name: str, *, required: bool = False) -> list[str]:
|
||||
if value is None:
|
||||
if required:
|
||||
raise ValidationError(f"{field_name} is required")
|
||||
return []
|
||||
if not isinstance(value, list):
|
||||
raise ValidationError(f"{field_name} must be an array of strings")
|
||||
result = []
|
||||
for item in value:
|
||||
text = str(item).strip()
|
||||
if text:
|
||||
result.append(text)
|
||||
if required and not result:
|
||||
raise ValidationError(f"{field_name} must contain at least one item")
|
||||
return result
|
||||
|
||||
|
||||
def _mapping(value: Any, field_name: str) -> dict[str, Any]:
|
||||
if not isinstance(value, dict):
|
||||
raise ValidationError(f"{field_name} must be an object")
|
||||
return value
|
||||
|
||||
|
||||
def _boolean(value: Any, field_name: str) -> bool:
|
||||
if not isinstance(value, bool):
|
||||
raise ValidationError(f"{field_name} must be a boolean")
|
||||
return value
|
||||
|
||||
|
||||
def _integer(value: Any, field_name: str) -> int:
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||
raise ValidationError(f"{field_name} must be an integer")
|
||||
if isinstance(value, float) and (not math.isfinite(value) or not value.is_integer()):
|
||||
raise ValidationError(f"{field_name} must be an integer")
|
||||
return int(value)
|
||||
|
||||
|
||||
def _optional_integer(value: Any, field_name: str) -> int | None:
|
||||
if value is None or value == "":
|
||||
return None
|
||||
return _integer(value, field_name)
|
||||
|
||||
|
||||
def _number(value: Any, field_name: str) -> float:
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||
raise ValidationError(f"{field_name} must be a finite number")
|
||||
result = float(value)
|
||||
if not math.isfinite(result):
|
||||
raise ValidationError(f"{field_name} must be a finite number")
|
||||
return result
|
||||
|
||||
|
||||
def unique_nonempty(values: Iterable[str]) -> list[str]:
|
||||
seen: set[str] = set()
|
||||
result: list[str] = []
|
||||
for value in values:
|
||||
text = value.strip()
|
||||
if text and text not in seen:
|
||||
seen.add(text)
|
||||
result.append(text)
|
||||
return result
|
||||
Reference in New Issue
Block a user