Phase 1 代码蓝图

本文档从代码维度描述 Phase 1 如何落地，但不编写生产代码。

它的作用是把 8 个 Phase 1 features 映射到具体模块、核心模型、接口和测试。

总体代码结构

src/
  kgent/
    __init__.py
    cli.py
    errors.py

    config/
      __init__.py
      models.py
      loader.py
      prompts.py

    runtime/
      __init__.py
      runner.py
      context.py
      options.py
      ids.py
      state.py

    engine/
      __init__.py
      base.py
      pydantic_ai.py
      mock.py

    sandbox/
      __init__.py
      workspace.py
      paths.py
      manifest.py

    records/
      __init__.py
      context.py
      events.py
      transcript.py
      logs.py
      errors.py

    skills/
      __init__.py
      models.py
      registry.py
      loader.py
      parser.py

    tools/
      __init__.py
      registry.py
      filesystem.py
      skills.py
      memory.py
      permissions.py
      results.py
      pydantic_adapter.py

    memory/
      __init__.py
      contract.py
      client.py

    governance/
      __init__.py
      deliverables.py
      policy.py
      results.py

tests/
  unit/
  integration/

模块职责

模块	职责	不负责
`cli.py`	解析 CLI 参数，调用 runtime。	不包含业务逻辑、工具逻辑或 Pydantic AI 细节。
`config/`	加载、校验、渲染配置和 profile。	不创建 run，不执行 agent。
`runtime/`	协调一次 run。	不硬编码领域知识，不直接实现工具。
`engine/`	隔离 Pydantic AI 调用。	不拥有 profile、skills、memory、sandbox、governance。
`sandbox/`	创建 run 工作区，解析路径，生成 manifests。	不执行模型或工具业务。
`records/`	写 events、transcript、logs。	不决定治理策略。
`skills/`	发现、索引、加载技能。	不授予工具权限。
`tools/`	暴露 agent 可调用工具，执行权限检查和日志。	不绕过 sandbox。
`memory/`	定义外部记忆 contract 和默认 client。	不内建记忆数据库。
`governance/`	最小校验、权限结果、交付物检查。	不做完整 reviewer agent。

核心数据模型

序列化模型策略

Phase 1 应明确哪些对象需要稳定序列化，哪些只是内部运行时辅助对象。

推荐策略：

类型	建议实现	原因
Config models	Pydantic models	需要严格校验 YAML 输入。
Event models	Pydantic models	需要稳定写入 JSONL。
Manifest models	Pydantic models	需要作为调度和环境重建契约。
External memory contracts	Pydantic models 或 Protocol + Pydantic DTO	需要和外部系统稳定交互。
Governance results	Pydantic models	需要写入 events 和 run state。
Internal runtime helpers	dataclass	只在进程内使用，保持轻量。
Engine request/result	Pydantic models 或 dataclass	若要写入日志或测试 snapshot，优先 Pydantic。

原则：

会写入文件、跨模块传递或作为外部契约的对象，优先使用 Pydantic。
只在函数内部或单模块内使用的辅助对象，可以使用 dataclass。
不要在 Phase 1 引入复杂 ORM 或数据库模型。

Config Models

位置：kgent/config/models.py

class KgentConfig:
    kgent: KgentMetaConfig
    model: ModelConfig
    profile: ProfileConfig
    skills: SkillsConfig
    tools: ToolsConfig
    memory: MemoryConfig
    workspace: WorkspaceConfig
    deliverables: DeliverablesConfig
    runtime: RuntimeConfig

关键要求：

kgent.schema_version 必填。
profile.id 必填且稳定。
memory.write_mode 默认 candidate。
tools.shell.enabled 默认 False。
tools.filesystem.delete 默认 False。
required deliverables 必须位于 deliverables/ 下。

CLI and Runtime Boundary Models

位置：kgent/runtime/options.py

class CliOverrides:
    sandbox: Path | None
    prompt: str | None
    prompt_file: Path | None
    session_id: str | None
    task_id: str | None
    run_id: str | None

class RunOptions:
    sandbox: Path | None
    session_id: str | None
    task_id: str | None
    run_id: str | None
    stream: bool | None
    max_steps: int | None
    timeout_seconds: int | None

class RunResult:
    session_id: str
    task_id: str
    run_id: str
    status: str
    sandbox_root: Path
    final_text: str | None
    error: ErrorInfo | None

关键要求：

CLI 只构造 CliOverrides 或 RunOptions，不直接操作 runtime internals。
CliOverrides、RunOptions、RunResult 统一位于 runtime/options.py。
RunResult 是 CLI 输出摘要的数据来源。
RunResult 不替代 run.json；run.json 仍是持久 source of truth。

Run Context

位置：kgent/runtime/context.py

class RunContext:
    session_id: str
    task_id: str
    run_id: str
    profile_id: str
    config_fingerprint: str
    sandbox_root: Path
    config: KgentConfig
    records: RunRecords

关键要求：

不包含模型内部状态。
可序列化为 run.json 的核心字段。
未来 scheduler 可基于该对象重建 run。

Manifest Models

位置：kgent/sandbox/manifest.py

class SandboxManifest:
    root: Path
    writable: list[str]
    readonly: list[str]
    forbidden: list[str]
    created_at: str

class ArtifactRecord:
    path: str
    kind: str
    created_by: str
    required: bool
    content_type: str | None

class ArtifactManifest:
    artifacts: list[ArtifactRecord]
    updated_at: str

关键要求：

sandbox-manifest.json Phase 1 必须生成。
artifact-manifest.json Phase 1 必须生成，可先只记录 required deliverables 和明确写入的 artifacts。
artifact paths 必须是 sandbox-relative paths。
manifests 是未来 scheduler 进行环境重建和 handoff 的基础。

Run State

位置：kgent/runtime/state.py

class RunStatus:
    pending
    running
    completed
    incomplete
    failed

Phase 1 只实现子集，但 schema 预留：

waiting_for_tool
waiting_for_user
review_required
revision_required
cancelled

Event

位置：kgent/records/events.py

class Event:
    event_id: str
    sequence: int
    run_id: str
    session_id: str
    task_id: str
    type: str
    timestamp: str
    actor: str
    severity: str
    summary: str
    data: dict
    correlation_id: str | None
    parent_event_id: str | None

关键要求：

sequence 是单个 run 内排序 source of truth。
写入 events.jsonl append-only。
data 不直接写入大对象或敏感值。

Skill Models

位置：kgent/skills/models.py

class SkillSummary:
    id: str
    name: str
    description: str
    tags: list[str]
    path: Path

class LoadedSkill:
    summary: SkillSummary
    instructions: str
    references_summary: str | None
    templates_summary: str | None
    declared_permissions: dict

关键要求：

discovery 只加载 metadata。
load 阶段有内容大小预算。
permissions 是声明，不是授权。

Tool Result

位置：kgent/tools/results.py

class ToolResult:
    ok: bool
    summary: str
    data: dict | None
    artifacts: list[str]
    error: ErrorInfo | None

关键要求：

大型输出写 artifact，不直接放入 data。
logs 使用 summary。
error 使用统一 ErrorInfo。

Tool Protocol

位置：kgent/tools/registry.py

class ToolDefinition:
    name: str
    description: str
    input_schema: dict
    handler: ToolHandler

class ToolHandler:
    def __call__(self, args: dict, context: RunContext) -> ToolResult:
        ...

关键要求：

build_toolsets 返回 kgent-native ToolDefinition 列表或等价 Protocol。
tools/pydantic_adapter.py 负责将 ToolDefinition 转换为 Pydantic AI 工具。
通用工具实现不直接依赖 Pydantic AI。

Tool Call Record

位置：kgent/records/logs.py

class ToolCallRecord:
    call_id: str
    tool_name: str
    action: str
    started_at: str
    completed_at: str | None
    duration_ms: int | None
    status: str
    args_summary: dict
    result_summary: str | None
    artifacts: list[str]
    error: ErrorInfo | None

关键要求：

每次 tool call 都有 call_id。
args_summary 不包含敏感值或大型内容。
大型结果写入 artifact，并通过 artifacts 引用。
ToolCallRecord 写入 logs/tools.jsonl。

Error Model

位置：kgent/errors.py

class ErrorInfo:
    code: str
    message: str
    category: str
    retryable: bool
    details: dict | None

建议 category：

config
sandbox
skill
tool
memory
engine
governance
unknown

关键要求：

AgentError、tool errors、governance failures 都应能转换为 ErrorInfo。
run.json.failure_reason 应引用 ErrorInfo.code 或摘要。
logs/errors.jsonl 应使用统一错误结构。
不将原始异常堆栈默认暴露给模型上下文。

ErrorInfo 是跨模块基础模型，不属于 records 模块，避免其他模块为了引用错误结构而依赖 records。

Run Records

位置：kgent/records/context.py

class RunRecords:
    events: EventRecorder
    transcript: TranscriptWriter
    tools: ToolLogger
    errors: ErrorLogger

关键要求：

Runtime 创建 RunRecords。
RunContext 持有 records: RunRecords。
下游模块不直接打开记录文件。

Governance Result

位置：kgent/governance/results.py

class GovernanceStatus:
    passed
    failed
    incomplete
    blocked

关键映射：

invalid config -> blocked
unsafe path -> blocked
missing required deliverable -> incomplete

核心接口设计

Config

位置：kgent/config/loader.py

def load_config(path: Path) -> KgentConfig:
    ...

def apply_cli_overrides(config: KgentConfig, overrides: CliOverrides) -> KgentConfig:
    ...

def fingerprint_config(config: KgentConfig) -> str:
    ...

位置：kgent/config/prompts.py

def render_system_prompt(
    config: KgentConfig,
    skill_index: list[SkillSummary],
    memory_summary: str | None,
    sandbox_rules: str,
) -> str:
    ...

Runtime

位置：kgent/runtime/runner.py

def run(config_path: Path, prompt: str, options: RunOptions) -> RunResult:
    ...

运行职责：

加载 config。
创建 IDs。
创建 sandbox。
初始化 records。
discover skills。
recall memory。
render prompt。
build tools。
invoke engine。
validate deliverables。
finalize run state。

Sandbox

位置：kgent/sandbox/workspace.py

def create_workspace(context: RunContext) -> SandboxWorkspace:
    ...

位置：kgent/sandbox/paths.py

def resolve_sandbox_path(root: Path, user_path: str, *, for_write: bool) -> Path:
    ...

必须拒绝：

绝对路径逃逸。
.. path traversal。
写入非允许目录。

Records

位置：kgent/records/events.py

class EventRecorder:
    def emit(self, event_type: str, summary: str, data: dict, *, actor: str, severity: str = "info") -> None:
        ...

位置：kgent/records/transcript.py

class TranscriptWriter:
    def append_section(self, title: str, content: str) -> None:
        ...
    def finalize(self, context: RunContext) -> None:
        ...

Skills

位置：kgent/skills/registry.py

class SkillRegistry:
    def discover(self) -> list[SkillSummary]:
        ...
    def load_skill(self, skill_id: str) -> LoadedSkill:
        ...

Tools

位置：kgent/tools/registry.py

def build_toolsets(context: RunContext, skills: SkillRegistry, memory: MemoryClient) -> list[ToolDefinition]:
    ...

位置：kgent/tools/permissions.py

def build_effective_policy(config: KgentConfig, loaded_skills: list[LoadedSkill]) -> EffectivePermissionPolicy:
    ...

def check_permission(policy: EffectivePermissionPolicy, tool_name: str, action: str, path: str | None = None) -> PermissionResult:
    ...

Memory

位置：kgent/memory/contract.py

class MemoryClient:
    def recall_memory(self, query: str, scope: str, limit: int) -> list[MemoryItem]:
        ...
    def write_memory(self, content: str, scope: str, tags: list[str]) -> MemoryWriteResult:
        ...

write_mode 行为：

disabled：不注册 write_memory 或返回 blocked。
candidate：写入 archive/candidate-memory.jsonl。
external：调用外部 memory client。

Engine

位置：kgent/engine/base.py

class AgentRunRequest:
    context: RunContext
    prompt: str
    system_prompt: str
    tools: list[ToolDefinition]
    timeout_seconds: int | None
    max_steps: int | None
    stream: bool
    cancellation_token: object | None
    event_callback: object | None

class AgentEngine:
    def run(self, request: AgentRunRequest) -> AgentResult:
        ...

AgentResult.error 应使用 ErrorInfo 或可转换为 ErrorInfo 的结构。

Pydantic AI 工具适配边界：

tools/
  定义 kgent-native tool handlers、ToolResult、PermissionPolicy。

tools/pydantic_adapter.py
  将 kgent tools 包装为 Pydantic AI 可用工具对象。

engine/
  消费 Pydantic AI compatible tools，但不实现工具业务逻辑。

通用 tools 不应直接依赖 Pydantic AI。

位置：kgent/engine/pydantic_ai.py

class PydanticAIAgentEngine(AgentEngine):
    def run(self, request: AgentRunRequest) -> AgentResult:
        ...

运行流

kgent run
  -> load_config
  -> apply_cli_overrides
  -> fingerprint_config
  -> create ids
  -> determine sandbox root
  -> create sandbox workspace
  -> init RunRecords
  -> create RunContext
  -> write run.json / config.yaml / prompt.md
  -> discover skills
  -> recall memory
  -> render effective system prompt
  -> build tools
  -> engine.run
  -> validate deliverables
  -> update run.json
  -> finalize transcript

测试策略

Unit Tests

tests/unit/config/

valid config loads。
missing profile role fails。
deliverables outside deliverables/ fails。
memory.write_mode default is candidate。
prompt composition order is deterministic。
config fingerprint is stable for equivalent resolved config。

tests/unit/sandbox/

create deterministic directory layout。
reject .. traversal。
reject absolute path outside root。
allow writes under workspace/ and deliverables/。
reject writes outside allowed dirs。
generate sandbox-manifest.json。
generate minimal artifact-manifest.json。

tests/unit/records/

event sequence increments。
JSONL append works。
event required fields exist。
large data is summarized。
transcript has minimal required sections。
RunRecords aggregates event/transcript/tool/error writers。

tests/unit/skills/

discover valid skill。
malformed front matter fails。
enabled missing skill fails。
loaded skill respects size budget。
skill permissions are declarations。

tests/unit/tools/

dangerous tools disabled by default。
effective permission policy deny rules win。
memory write candidate writes archive file。
tool result summary contract is honored。
tool call record includes call id, duration, status, summaries, and artifacts。

tests/unit/governance/

missing deliverable -> incomplete。
invalid config -> blocked。
unsafe path -> blocked。

tests/unit/engine/

engine can be mocked。
structured error conversion works。
request includes timeout/max_steps/cancellation/event_callback。
engine failure path updates run result with ErrorInfo。

tests/unit/cli/

CLI parses --config and --prompt。
CLI parses --prompt-file。
CLI applies --sandbox override。
CLI forwards session_id、task_id、run_id into RunOptions。

Integration Tests

tests/integration/test_minimal_run.py

目标：不用真实模型，使用 mock engine 完成一次最小 run。

验证：

sandbox 被创建。
config/prompt/effective prompt 被写入。
events/transcript 被写入。
skill index 被构建。
required deliverables 被检查。
run state 最终为 completed 或 incomplete。

tests/integration/test_filesystem_safety.py

验证：

agent/tool 无法读写 sandbox 外路径。
path rejection 写入 events 和 errors。

tests/integration/test_memory_candidate_mode.py

验证：

memory.write_mode: candidate 时写入 archive/candidate-memory.jsonl。

tests/integration/test_engine_failure_path.py

验证：

mock engine failure 写入 events.jsonl。
logs/errors.jsonl 使用 ErrorInfo。
run.json.status 变为 failed。

tests/integration/test_manifest_outputs.py

验证：

run 结束后存在 sandbox-manifest.json。
run 结束后存在 artifact-manifest.json。
artifact paths 都是 sandbox-relative paths。

实现顺序建议

pyproject.toml、package skeleton、CLI 空命令。
config models、loader、prompt renderer。
CliOverrides、RunOptions、RunResult。
ids、RunContext、run state。
sandbox workspace、path guards、manifests。
ErrorInfo、EventRecorder、TranscriptWriter。
SkillRegistry。
ToolResult、PermissionPolicy、filesystem tools、Pydantic tool adapter。
Memory contract 和 candidate write mode。
AgentEngine base 和 mock engine。
Pydantic AI adapter。
governance deliverable checks。
integration minimal run。

不变量

后续代码实现必须保持：

Runtime 不硬编码专业领域。
Model 不获得不受限制的 filesystem access。
Skills 不直接授予权限。
Memory 不内建数据库。
Events append-only。
run.json 是 Phase 1 run state source of truth。
incomplete 是 Phase 1 RunStatus。
sandbox-manifest.json 和 artifact-manifest.json 必须生成。
大型输出写 artifact，不直接塞入 event data。
统一错误结构使用 ErrorInfo。
ErrorInfo 统一位于 kgent/errors.py。
ToolCallRecord 统一位于 records/logs.py。
所有可替换高成本能力都位于接口后。

来源项目	`kunora-kgent`
分支	`docs-publish`
路径	`technology/components/kunora-kgent/code-design/phase1-code-blueprint.md`

总体代码结构​

模块职责​

核心数据模型​

序列化模型策略​

Config Models​

CLI and Runtime Boundary Models​

Run Context​

Manifest Models​

Run State​

Event​

Skill Models​

Tool Result​

Tool Protocol​

Tool Call Record​

Error Model​

Run Records​

Governance Result​

核心接口设计​

Config​

Runtime​

Sandbox​

Records​

Skills​

Tools​

Memory​

Engine​

运行流​

测试策略​

Unit Tests​

Integration Tests​

实现顺序建议​

不变量​

总体代码结构

模块职责

核心数据模型

序列化模型策略

Config Models

CLI and Runtime Boundary Models

Run Context

Manifest Models

Run State

Event

Skill Models

Tool Result

Tool Protocol

Tool Call Record

Error Model

Run Records

Governance Result

核心接口设计

Config

Runtime

Sandbox

Records

Skills

Tools

Memory

Engine

运行流

测试策略

Unit Tests

Integration Tests

实现顺序建议

不变量