Abstract First Coder

Abstract First Coder

This skill guides agents to write code in the style observed in /Users/ruosongchen/PyCharmMiscProject/knowledge_graph_xm: everything starts from a stable abstraction, then concrete implementations are plugged into a pipeline. If that path is unreachable, or the repo you are in shows a different style, take the row beginning The reference project named in this skill's opening paragraph is unavailable in Execution Failures And Recovery before following the style above.

Core Principle

Do not start by writing a one-off function that directly solves the current case. First identify the concept, define its contract, then implement the current strategy behind that contract.

Preferred shape:

1. dataclass / BaseModel defines the data.
2. ABC interface defines the capability.
3. Concrete classes implement strategies.
4. Default implementation provides a safe baseline.
5. Pipeline or service class orchestrates injected capabilities.
6. Tests or single-point checks verify each unit and the composed flow.

Scale Abstraction To Task Size

Abstraction is a tool for anticipated variation and replacement, not a tax paid on every task. Match the depth of abstraction to how likely the code is to change, vary, or be swapped. Over-abstracting a one-off is as much a defect as under-abstracting a core capability.

Decide depth before coding, using this ladder:

• One-off, single call site, no foreseeable variant → a plain function or a passive dataclass is the correct and complete abstraction. Do not add an ABC, strategy class, or Default fallback. A two-function module is a valid end state.
• One concept, but a second strategy is plausible soon, or it crosses a domain boundary → define the dataclass + a thin ABC, implement the single concrete strategy now. Defer Default/injection until the second strategy actually appears.
• Core capability, replaceable, multiple strategies, or an agent/human/code decision point → full shape: data model + interface + strategies + default + injection. This is where the layering below earns its cost.

Guardrails so the abstraction-first habit does not become ceremony:

• Every interface, strategy class, or injection point must name the concrete second case or replacement it exists to enable. If you cannot name one, collapse it to a function until one exists.
• Do not build configurability, extension hooks, or error handling for scenarios that cannot currently occur.
• Prefer the smallest shape that still makes the abstraction boundary obvious; expand it the moment a real second case arrives, not before.
• Self-check: would a senior engineer call this structure premature for the task as scoped? If yes, drop a level on the ladder.

When in doubt for substantial, long-lived, or pipeline code, lean toward the abstraction. For scripts, glue, single-use transforms, and small features, lean toward the plain function. State which rung you chose and why before implementing. If the rung turns out too high once built, take the row beginning Abstraction level turns out too high in Execution Failures And Recovery; if the user disagrees with it, take the row beginning You and the user disagree on task scale.

Architecture Pattern

Use this layering unless the repo already has a stronger local convention. When the repo's convention conflicts with it, take the row beginning Repo has a stronger local convention in Execution Failures And Recovery rather than choosing on the user's behalf.

*_definitions/         domain data structures and abstract processors
ability_ext/           generic capability interfaces and code-side adapters
data_process/          domain processing strategies
utils/                 infrastructure adapters and reusable utilities
pipeline.py            orchestration only, minimal business details
scripts/               executable entrypoints
single_points_test/    focused checks for individual abilities

Naming Pattern

Use names that reveal the abstraction boundary.

• Abstract capability: Classification, Pile, Chunker, Download, LLM, GraphResult.
• Concrete strategy: ClassificationByDefault, PileByInfoDensity, ChunkTextByRecursive, DownloadTextFiles.
• Default fallback: XDefault or DefaultX when it is a safe baseline.
• Data definition: noun-centered names such as InputData, Document, KGTriple, TriplesForChecking.
• Capability adapter: CodeFilter, CodeFillForm, CodeDecision, CodeKGExtractionDecision.
• Pipeline method: verb phrase for one stage, such as prepare_run_context, prepare_sample_chunks, run_recursive_extraction.

Required Design Pass Before Coding

Before implementing a new feature, write or mentally settle these answers:

• What is the domain object? Define it first.
• What is the capability being added? Define an abstract interface.
• Which concrete strategy is needed now?
• What is the default or fallback behavior?
• Where will the pipeline inject and call it?
• What data enters and leaves the capability?
• What errors should be raised vs tolerated?
• What focused verification proves the capability works?

🛑 STOP — do not write implementation code while any answer above is unsettled. A contract guessed now is redefined later at the cost of every caller already written against it. Resolve it, or ask the user.

Data First

Define data contracts close to the domain.

Use dataclass for simple internal records:

from dataclasses import dataclass

@dataclass
class Document:
    link: str
    local_path: str | None = None

Use BaseModel when validation, schema metadata, or structured LLM/tool input matters.

Keep data classes mostly passive. Put transformation logic in processors, strategies, or pipeline stages.

Interface First

Every replaceable capability should have an ABC.

from abc import ABC, abstractmethod

class Chunker(ABC):
    @abstractmethod
    async def chunk(self, text: str) -> list[str]:
        pass

Then add concrete strategies:

class ChunkTextByRecursive(Chunker):
    async def chunk(self, text: str) -> list[str]:
        ...

class ChunkerDefault(Chunker):
    def __init__(self, length: int = 512, overlap: int = 0):
        self.chunker = ChunkTextByRecursive(length, overlap)

    async def chunk(self, text: str) -> list[str]:
        return await self.chunker.chunk(text)

Capability Injection

Prefer passing implementations into pipeline methods instead of hard-coding behavior inside the orchestration.

Good:

chunked_result = await self.chunk(run_context["chunker"], documents)

Avoid:

chunker = ChunkTextByRecursive(...)
chunked_result = await chunker.chunk(documents)

Creating defaults in prepare_run_context is acceptable because it centralizes runtime choices. If a capability that must be injected has no implementation available at the call site, take the row beginning A capability that must be injected is unavailable in Execution Failures And Recovery; never reach for a hard-coded internal instance.

Pipeline Style

Pipeline methods should read like a workflow and delegate details.

Good orchestration:

run_context = await self.prepare_run_context(input_datas)
await self.prepare_and_confirm_template(input_datas)
pile_result = await self.prepare_documents_and_piles(input_datas, run_context)
sample_chunks = await self.prepare_sample_chunks(input_datas, pile_result, run_context)

Each stage should:

• Accept explicit inputs.
• Return explicit outputs.
• Call abstract capabilities or helper methods.
• Record important intermediate artifacts when the project expects traceability.
• Keep low-level parsing, scoring, IO, and formatting in helpers or strategy classes.

Async Contract

If a capability may call IO, LLMs, network, files, user/agent review, or long-running work, make the interface async from the beginning. Keep async signatures consistent across abstract and concrete classes.

Use small helpers such as _await_if_needed only at adapter boundaries where both sync and async handlers are intentionally supported.

Strategy Classes

When adding a new algorithm, do not replace the old one in place. Add a new strategy class under the same abstraction.

Example pattern:

class Pile(ABC):
    @abstractmethod
    async def pile(self, sampling_accuracy, input_datas):
        pass

class PileByInfoDensity(Pile):
    async def pile(self, sampling_accuracy, input_datas):
        ...

class PileByDefault(Pile):
    async def pile(self, sampling_accuracy, input_datas):
        ...

This lets the pipeline choose strategies without changing downstream contracts. If the contract turns out to force callers to pass irrelevant arguments, take the row beginning Contract does not hold once you write the implementation in Execution Failures And Recovery.

Defaults And Fallbacks

Every important capability should have a baseline:

• Default returns simple deterministic behavior.
• Code catches optional integration failures only where fallback is intentional.
• Fallbacks must preserve the interface contract.
• If fallback may reduce quality, log or record that fact.

Do not silently swallow errors in core correctness paths.

Agent Ability Pattern

For agent/human/code decision points, separate the ability interface from the code-side implementation.

Use generic capability interfaces:

• IFilter[T]
• IFillForm[T]
• IDecision[I, O]
• IJudgingOption[T]

Then adapt code functions into those interfaces:

• CodeFilter
• CodeFillForm
• CodeDecision
• CodeJudgingOption

This makes later replacement by UI, human review, tool calls, or model decisions straightforward.

Error Handling And Logging

• Wrap pipeline-stage errors with stage/function context.
• Log retries and final failures for external calls.
• Raise meaningful errors when required capabilities are not injected.
• Validate input contracts early with check_* functions.
• Keep recoverable fallback behavior local and explicit.

File And Artifact Discipline

For data pipelines, write intermediate artifacts when they help debugging or review:

• classification result
• sample chunks and similarity scores
• template change history
• extraction history
• metrics
• final result

Generated files should live under a run-specific temp/output path instead of polluting source directories.

Implementation Checklist

🔴 CHECKPOINT — run this list against the code that actually exists, not against the plan. A structure that satisfies the list only in intent does not satisfy it.

• A domain data model exists for new structured data.
• A replaceable capability has an abstract interface.
• The concrete implementation satisfies the interface.
• A default/fallback exists when appropriate.
• Pipeline orchestration is readable and not overloaded with low-level details.
• Dependencies are injected or centralized in a run context.
• Errors and retries are handled at the correct boundary.
• Inputs are validated before processing.
• Outputs are saved or returned in a traceable shape.
• A focused test, script, or single-point check exists for the new capability.

Execution Failures And Recovery

This skill itself goes wrong in predictable ways. Handle by table, never continue silently.

What To Avoid

• Large procedural functions that mix input validation, algorithm logic, IO, LLM calls, persistence, and formatting.
• Hard-coded concrete classes scattered across many pipeline stages.
• Adding behavior directly to a data class.
• Changing an existing strategy in a way that removes previous behavior without adding a new strategy or compatibility path.
• Sync interfaces for capabilities likely to require async IO later.
• Unnamed dictionaries where a dataclass or model would clarify the contract.
• Hidden global state for pipeline decisions.

Minimal Example Shape

from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class ScoreInput:
    text: str
    topic: str

@dataclass
class ScoreResult:
    value: float
    reason: str

class Scorer(ABC):
    @abstractmethod
    async def score(self, source: ScoreInput) -> ScoreResult:
        pass

class ScorerByKeywordDensity(Scorer):
    async def score(self, source: ScoreInput) -> ScoreResult:
        value = source.text.count(source.topic) / max(1, len(source.text))
        return ScoreResult(value=value, reason="keyword_density")

class ScorerDefault(Scorer):
    async def score(self, source: ScoreInput) -> ScoreResult:
        return ScoreResult(value=0.0, reason="default")

Use this shape as the default starting point when adding new capabilities.