教程概述
系列目录：CLI-Anything 教程系列索引
Implement 是 CLI-Anything 7 阶段流程的第三阶段。在这一阶段，我们将把 Design 阶段的架构设计转化为可运行的代码。
你将学到
- ✅ CLI 项目结构
- ✅ Click 框架核心用法
- ✅ REPL 实现技巧
- ✅ 状态管理机制
- ✅ 错误处理和日志
项目结构
标准目录布局
<software>/agent-harness/
├── cli_anything/
│ ├── __init__.py # 包初始化
│ ├── cli.py # 主 CLI 入口
│ ├── repl.py # REPL 实现
│ ├── state.py # 状态管理
│ ├── core/ # 核心功能模块
│ │ ├── __init__.py
│ │ ├── project.py # 项目管理
│ │ ├── image.py # 图像操作
│ │ ├── layer.py # 图层操作
│ │ └── ...
│ ├── utils/ # 工具函数
│ │ ├── __init__.py
│ │ ├── backend.py # 后端软件封装
│ │ ├── validators.py # 参数验证
│ │ └── output.py # 输出格式化
│ └── tests/ # 测试目录
│ ├── __init__.py
│ ├── unit/ # 单元测试
│ └── e2e/ # E2E 测试
├── setup.py # 安装配置
├── README.md # 使用说明
└── SKILL.md # AI 技能定义
Click 框架基础
什么是 Click？
Click 是一个 Python 的命令行界面创建工具包，用于快速构建漂亮的 CLI。
import click
@click.command()
@click.option('--count', default=1, help='Number of greetings.')
@click.option('--name', prompt='Your name', help='The person to greet.')
def hello(count, name):
"""Simple program that greets NAME for a total of COUNT times."""
for _ in range(count):
click.echo(f"Hello {name}!")
if __name__ == '__main__':
hello()
核心概念
flowchart TD
A[Click 核心概念] --> B[Command 命令]
A --> C[Group 命令组]
A --> D[Option 选项]
A --> E[Argument 参数]
B --> F[@click.command]
C --> G[@click.group]
D --> H[@click.option]
E --> I[@click.argument]
CLI 入口实现
主 CLI 文件 (cli.py)
#!/usr/bin/env python3
"""
cli-anything-gimp - GIMP CLI for AI Agents
A command-line interface for GIMP that enables AI agents to
automate image editing workflows.
"""
import click
import json
import sys
from pathlib import Path
from typing import Optional
from .repl import start_repl
from .state import ProjectState, StateManager
from .core.project import project_group
from .core.image import image_group
from .core.layer import layer_group
from .core.filter import filter_group
from .core.export import export_group
from .utils.output import format_output, OutputFormat
# 版本信息
__version__ = "1.0.0"
@click.group()
@click.version_option(version=__version__)
@click.option('--json', 'json_output', is_flag=True,
help='Output in JSON format for agent consumption')
@click.option('--project', 'project_path', type=click.Path(),
help='Path to project state file')
@click.option('--verbose', is_flag=True, help='Enable verbose output')
@click.option('--quiet', is_flag=True, help='Suppress non-error output')
@click.pass_context
def cli(ctx: click.Context, json_output: bool, project_path: Optional[str],
verbose: bool, quiet: bool):
"""GIMP CLI for AI Agents - Automate image editing workflows.
Examples:
cli-anything-gimp # Start REPL mode
cli-anything-gimp project new # Create new project
cli-anything-gimp --json image info # Get image info as JSON
"""
# 确保 ctx.obj 存在
if ctx.obj is None:
ctx.obj = {}
# 存储全局选项
ctx.obj['json_output'] = json_output
ctx.obj['verbose'] = verbose
ctx.obj['quiet'] = quiet
# 初始化状态管理器
ctx.obj['state_manager'] = StateManager()
# 加载项目状态（如果指定）
if project_path:
try:
state = ctx.obj['state_manager'].load(project_path)
ctx.obj['project_state'] = state
ctx.obj['project_path'] = project_path
except Exception as e:
if not quiet:
click.echo(f"Error loading project: {e}", err=True)
sys.exit(1)
# 注册命令组
cli.add_command(project_group)
cli.add_command(image_group)
cli.add_command(layer_group)
cli.add_command(filter_group)
cli.add_command(export_group)
@cli.command()
def repl():
"""Start interactive REPL mode."""
start_repl()
def main():
"""Entry point for the CLI."""
cli()
if __name__ == '__main__':
main()
命令组实现
# cli_anything/core/project.py
import click
from pathlib import Path
from datetime import datetime
from ..state import ProjectState
from ..utils.output import success, error, info
@click.group(name='project')
@click.pass_context
def project_group(ctx):
"""Project management commands."""
pass
@project_group.command(name='new')
@click.option('--name', '-n', required=True, help='Project name')
@click.option('--template', '-t', default='default',
type=click.Choice(['default', 'photo', 'web', 'print']),
help='Project template')
@click.option('--output', '-o', type=click.Path(),
help='Output path for project state file')
@click.pass_context
def project_new(ctx, name: str, template: str, output: Optional[str]):
"""Create a new project.
Examples:
cli-anything-gimp project new --name "MyProject"
cli-anything-gimp project new -n "PhotoEdit" -t photo
"""
try:
# 创建项目状态
state = ProjectState(name=name, template=template)
# 确定状态文件路径
if output:
state_path = Path(output)
else:
state_path = Path(f"{name.lower().replace(' ', '_')}.json")
# 保存状态
state.save(state_path)
# 输出结果
result = {
'project_id': state.project_id,
'project_name': state.name,
'template': template,
'state_file': str(state_path)
}
if ctx.obj.get('json_output'):
click.echo(json.dumps({
'success': True,
'command': 'project new',
'data': result
}))
else:
success(f"Created project: {name}")
info(f"Project ID: {state.project_id}")
info(f"Template: {template}")
info(f"State file: {state_path}")
# 更新上下文
ctx.obj['project_state'] = state
ctx.obj['project_path'] = str(state_path)
except Exception as e:
if ctx.obj.get('json_output'):
click.echo(json.dumps({
'success': False,
'command': 'project new',
'error': {
'type': type(e).__name__,
'message': str(e)
}
}))
else:
error(f"Failed to create project: {e}")
raise click.ClickException(str(e))
@project_group.command(name='info')
@click.pass_context
def project_info(ctx):
"""Show project information."""
state = ctx.obj.get('project_state')
if not state:
error("No project loaded. Use --project or run 'project new'")
raise click.ClickException("No project loaded")
info_data = {
'project_id': state.project_id,
'name': state.name,
'template': state.template,
'created_at': state.created_at.isoformat(),
'modified_at': state.modified_at.isoformat(),
'image_count': len(state.images),
'modified': state.modified
}
if ctx.obj.get('json_output'):
click.echo(json.dumps({
'success': True,
'command': 'project info',
'data': info_data
}))
else:
info("Project Information:")
for key, value in info_data.items():
click.echo(f" {key}: {value}")
REPL 实现
REPL 核心代码
# cli_anything/repl.py
import cmd
import shlex
import sys
from pathlib import Path
from typing import Optional
import click
from click.testing import CliRunner
from .cli import cli
from .state import ProjectState, StateManager
class CLIRepl(cmd.Cmd):
"""Interactive REPL for CLI-Anything."""
intro = '''
╔══════════════════════════════════════════╗
║ cli-anything-gimp v1.0.0 ║
║ GIMP CLI for AI Agents ║
╚══════════════════════════════════════════╝
Type 'help' for available commands, 'exit' to quit.
'''
def __init__(self):
super().__init__()
self.state_manager = StateManager()
self.project_state: Optional[ProjectState] = None
self.project_path: Optional[str] = None
self.history: list = []
self._update_prompt()
def _update_prompt(self):
"""Update the REPL prompt based on current state."""
base = "gimp"
if self.project_state:
base += f"[{self.project_state.name}]"
if self.project_state.modified:
base += "*"
self.prompt = f"{base}> "
def default(self, line: str):
"""Handle unknown commands by passing to Click CLI."""
if not line.strip():
return
# 解析命令
args = shlex.split(line)
# 记录历史
self.history.append(line)
# 使用 Click 测试运行器执行命令
runner = CliRunner()
# 构建参数
invoke_args = ['--json'] # REPL 模式下默认 JSON 输出
if self.project_path:
invoke_args.extend(['--project', self.project_path])
invoke_args.extend(args)
# 执行命令
result = runner.invoke(cli, invoke_args)
# 输出结果
if result.output:
print(result.output.strip())
# 处理错误
if result.exit_code != 0 and result.exception:
print(f"Error: {result.exception}", file=sys.stderr)
# 更新状态（如果项目被修改）
if self.project_path:
try:
self.project_state = self.state_manager.load(self.project_path)
self._update_prompt()
except:
pass
def do_help(self, arg: str):
"""Show help information."""
if arg:
# 显示特定命令的帮助
runner = CliRunner()
result = runner.invoke(cli, ['--help', arg])
print(result.output)
else:
# 显示所有命令
print("\nAvailable commands:")
print(" project Project management")
print(" image Image operations")
print(" layer Layer operations")
print(" filter Filters and effects")
print(" export Export and rendering")
print("\nBuilt-in commands:")
print(" help Show this help")
print(" history Show command history")
print(" save Save current state")
print(" status Show current status")
print(" undo Undo last operation")
print(" redo Redo last operation")
print(" exit Exit REPL")
def do_history(self, arg: str):
"""Show command history."""
for i, cmd in enumerate(self.history, 1):
print(f"{i:3d}: {cmd}")
def do_save(self, arg: str):
"""Save current project state."""
if not self.project_state:
print("No project to save")
return
try:
self.project_state.save(self.project_path)
self.project_state.modified = False
self._update_prompt()
print(f"✓ Saved project state to {self.project_path}")
except Exception as e:
print(f"✗ Failed to save: {e}")
def do_status(self, arg: str):
"""Show current status."""
if self.project_state:
print(f"Project: {self.project_state.name}")
print(f" ID: {self.project_state.project_id}")
print(f" Modified: {self.project_state.modified}")
print(f" Images: {len(self.project_state.images)}")
else:
print("No project loaded")
def do_undo(self, arg: str):
"""Undo last operation."""
if not self.project_state:
print("No project to undo")
return
# 实现撤销逻辑
if self.project_state.undo():
self._update_prompt()
print("✓ Undone last operation")
else:
print("Nothing to undo")
def do_redo(self, arg: str):
"""Redo last undone operation."""
if not self.project_state:
print("No project to redo")
return
# 实现重做逻辑
if self.project_state.redo():
self._update_prompt()
print("✓ Redone last operation")
else:
print("Nothing to redo")
def do_exit(self, arg: str):
"""Exit REPL."""
# 检查是否有未保存的更改
if self.project_state and self.project_state.modified:
response = input("You have unsaved changes. Save before exit? [Y/n] ")
if response.lower() not in ('n', 'no'):
self.do_save("")
print("Goodbye! 👋")
return True
do_quit = do_exit
do_EOF = do_exit # Ctrl+D
def start_repl():
"""Start the interactive REPL."""
try:
import readline
# 启用历史记录
readline.parse_and_bind('tab: complete')
except ImportError:
pass
repl = CLIRepl()
repl.cmdloop()
状态管理实现
状态管理核心
# cli_anything/state.py
import json
import uuid
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field, asdict
@dataclass
class LayerState:
"""Represents a layer state."""
layer_id: str
name: str
visible: bool = True
opacity: float = 100.0
position: tuple = field(default_factory=lambda: (0, 0))
def to_dict(self) -> dict:
return {
'layer_id': self.layer_id,
'name': self.name,
'visible': self.visible,
'opacity': self.opacity,
'position': self.position
}
@classmethod
def from_dict(cls, data: dict) -> 'LayerState':
return cls(**data)
@dataclass
class ImageState:
"""Represents an image state."""
image_id: str
name: str
width: int
height: int
layers: List[LayerState] = field(default_factory=list)
active_layer_id: Optional[str] = None
def to_dict(self) -> dict:
return {
'image_id': self.image_id,
'name': self.name,
'width': self.width,
'height': self.height,
'layers': [layer.to_dict() for layer in self.layers],
'active_layer_id': self.active_layer_id
}
@classmethod
def from_dict(cls, data: dict) -> 'ImageState':
layers = [LayerState.from_dict(l) for l in data.get('layers', [])]
return cls(
image_id=data['image_id'],
name=data['name'],
width=data['width'],
height=data['height'],
layers=layers,
active_layer_id=data.get('active_layer_id')
)
@dataclass
class Command:
"""Represents a command in history."""
command: str
args: dict
timestamp: datetime
undo_data: Optional[dict] = None
def to_dict(self) -> dict:
return {
'command': self.command,
'args': self.args,
'timestamp': self.timestamp.isoformat(),
'undo_data': self.undo_data
}
@dataclass
class ProjectState:
"""Represents the complete project state."""
name: str
template: str = 'default'
project_id: str = field(default_factory=lambda: f"proj_{uuid.uuid4().hex[:8]}")
created_at: datetime = field(default_factory=datetime.utcnow)
modified_at: datetime = field(default_factory=datetime.utcnow)
images: Dict[str, ImageState] = field(default_factory=dict)
active_image_id: Optional[str] = None
modified: bool = False
history: List[Command] = field(default_factory=list)
history_position: int = -1
def to_dict(self) -> dict:
return {
'version': '1.0.0',
'project_id': self.project_id,
'name': self.name,
'template': self.template,
'created_at': self.created_at.isoformat(),
'modified_at': self.modified_at.isoformat(),
'images': {k: v.to_dict() for k, v in self.images.items()},
'active_image_id': self.active_image_id,
'modified': self.modified,
'history': [cmd.to_dict() for cmd in self.history],
'history_position': self.history_position
}
def save(self, path: Path):
"""Save state to file."""
self.modified_at = datetime.utcnow()
with open(path, 'w') as f:
json.dump(self.to_dict(), f, indent=2)
self.modified = False
@classmethod
def load(cls, path: Path) -> 'ProjectState':
"""Load state from file."""
with open(path, 'r') as f:
data = json.load(f)
images = {k: ImageState.from_dict(v) for k, v in data.get('images', {}).items()}
history = [
Command(
cmd=cmd['command'],
args=cmd['args'],
timestamp=datetime.fromisoformat(cmd['timestamp']),
undo_data=cmd.get('undo_data')
)
for cmd in data.get('history', [])
]
return cls(
name=data['name'],
template=data.get('template', 'default'),
project_id=data['project_id'],
created_at=datetime.fromisoformat(data['created_at']),
modified_at=datetime.fromisoformat(data['modified_at']),
images=images,
active_image_id=data.get('active_image_id'),
modified=data.get('modified', False),
history=history,
history_position=data.get('history_position', -1)
)
def add_command(self, command: str, args: dict, undo_data: Optional[dict] = None):
"""Add a command to history."""
# 如果不在历史末尾，截断后续历史
if self.history_position < len(self.history) - 1:
self.history = self.history[:self.history_position + 1]
cmd = Command(
command=command,
args=args,
timestamp=datetime.utcnow(),
undo_data=undo_data
)
self.history.append(cmd)
self.history_position += 1
self.modified = True
def undo(self) -> bool:
"""Undo last operation."""
if self.history_position < 0:
return False
# 获取当前命令
cmd = self.history[self.history_position]
# 执行撤销（使用保存的 undo_data）
if cmd.undo_data:
self._apply_undo(cmd.undo_data)
self.history_position -= 1
self.modified = True
return True
def redo(self) -> bool:
"""Redo last undone operation."""
if self.history_position >= len(self.history) - 1:
return False
self.history_position += 1
cmd = self.history[self.history_position]
# 重新执行命令
# 实际实现需要调用相应的处理函数
self.modified = True
return True
def _apply_undo(self, undo_data: dict):
"""Apply undo data to restore state."""
# 根据 undo_data 恢复状态
pass
class StateManager:
"""Manages project state persistence."""
def load(self, path: str) -> ProjectState:
"""Load project state from file."""
return ProjectState.load(Path(path))
def save(self, state: ProjectState, path: str):
"""Save project state to file."""
state.save(Path(path))
小结
Implement 阶段将设计转化为可运行的代码：
- 项目结构 - 标准化的目录布局
- Click CLI - 命令组和选项的实现
- REPL 环境 - 交互式会话支持
- 状态管理 - 状态持久化和撤销/重做
关键要点
- ✅ 使用 Click 的标准模式
- ✅ 支持 JSON 和人类可读两种输出
- ✅ REPL 和子命令模式一致
- ✅ 完善的错误处理和日志
- ✅ 状态管理支持会话恢复
系列导航：
- ← 上一篇：教程 3：第二阶段 - 架构设计
- → 下一篇：教程 5：第四阶段 - 测试计划
- 返回：教程系列索引