feat: stage-1 complete - 最小 agentic loop + Harness + 3 tools (非流式)

2026-04-10 17:26:44 +08:00
parent 5528c0b3ca
commit 54191f8458
10 changed files with 1119 additions and 14 deletions
@@ -0,0 +1,3 @@
+__all__ = ["__version__"]
+
+__version__ = "0.1.0"
@@ -0,0 +1,268 @@
+from __future__ import annotations
+
+import json
+import os
+import tomllib
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from anthropic import Anthropic
+from openai import OpenAI
+
+from cc_slim.tools import Tool
+
+
+@dataclass
+class Config:
+    provider: str
+    model: str
+    api_key: str
+    base_url: str | None
+    max_turns: int = 8
+
+
+def resolve_config(workspace: Path, cli: dict[str, Any]) -> Config:
+    file_cfg = _load_file_config(workspace / ".cc-slim.toml")
+    provider = _pick(cli.get("provider"), os.getenv("CC_SLIM_PROVIDER"), file_cfg.get("provider"), "openai")
+    model = _pick(
+        cli.get("model"),
+        os.getenv("CC_SLIM_MODEL"),
+        file_cfg.get("model"),
+        "gpt-4.1-mini" if provider == "openai" else "claude-3-5-haiku-latest",
+    )
+    api_key = _pick(
+        cli.get("api_key"),
+        os.getenv("CC_SLIM_API_KEY"),
+        os.getenv("OPENAI_API_KEY") if provider == "openai" else os.getenv("ANTHROPIC_API_KEY"),
+        file_cfg.get("api_key"),
+        "",
+    )
+    base_url = _pick(cli.get("base_url"), os.getenv("CC_SLIM_BASE_URL"), file_cfg.get("base_url"), None)
+    max_turns_raw = _pick(cli.get("max_turns"), os.getenv("CC_SLIM_MAX_TURNS"), file_cfg.get("max_turns"), 8)
+
+    if not api_key:
+        raise ValueError("缺少 API key，请通过 CLI、环境变量或 .cc-slim.toml 提供。")
+
+    return Config(
+        provider=str(provider).strip().lower(),
+        model=str(model).strip(),
+        api_key=str(api_key).strip(),
+        base_url=str(base_url).strip() if base_url else None,
+        max_turns=int(max_turns_raw),
+    )
+
+
+class Agent:
+    def __init__(self, config: Config, tools: list[Tool], workspace: Path) -> None:
+        self.config = config
+        self.tools = {tool.name: tool for tool in tools}
+        self.history: list[dict[str, Any]] = []
+        self.system_prompt = self._build_system_prompt(workspace)
+        self.client = self._build_client()
+
+    def reply(self, user_input: str) -> str:
+        self.history.append({"role": "user", "content": user_input})
+
+        for _ in range(self.config.max_turns):
+            result = self._call_model()
+            self.history.append(result["assistant"])
+
+            if not result["tool_calls"]:
+                return result["text"].strip() or "(empty response)"
+
+            for call in result["tool_calls"]:
+                tool_output = self._run_tool(call["name"], call["input"])
+                self.history.append(
+                    {
+                        "role": "tool",
+                        "tool_call_id": call["id"],
+                        "name": call["name"],
+                        "content": tool_output,
+                    }
+                )
+
+        return "已达到最大工具循环轮数，停止执行。"
+
+    def _build_client(self) -> Any:
+        if self.config.provider == "openai":
+            kwargs: dict[str, Any] = {"api_key": self.config.api_key}
+            if self.config.base_url:
+                kwargs["base_url"] = self.config.base_url
+            return OpenAI(**kwargs)
+        if self.config.provider == "anthropic":
+            kwargs = {"api_key": self.config.api_key}
+            if self.config.base_url:
+                kwargs["base_url"] = self.config.base_url
+            return Anthropic(**kwargs)
+        raise ValueError(f"不支持的 provider: {self.config.provider}")
+
+    def _build_system_prompt(self, workspace: Path) -> str:
+        parts: list[str] = []
+        agents = workspace / "AGENTS.md"
+        if agents.exists():
+            parts.append(agents.read_text(encoding="utf-8"))
+
+        skills_dir = workspace / "SKILLS"
+        if skills_dir.exists():
+            for path in sorted(skills_dir.glob("*.md"), key=lambda p: p.name):
+                parts.append(path.read_text(encoding="utf-8"))
+
+        return "\n\n".join(part.strip() for part in parts if part.strip())
+
+    def _call_model(self) -> dict[str, Any]:
+        if self.config.provider == "openai":
+            return self._call_openai()
+        return self._call_anthropic()
+
+    def _call_openai(self) -> dict[str, Any]:
+        response = self.client.chat.completions.create(
+            model=self.config.model,
+            messages=self._openai_messages(),
+            tools=self._openai_tools(),
+            tool_choice="auto",
+        )
+        message = response.choices[0].message
+        text = message.content or ""
+        tool_calls = []
+        for call in message.tool_calls or []:
+            tool_calls.append(
+                {
+                    "id": call.id,
+                    "name": call.function.name,
+                    "input": json.loads(call.function.arguments or "{}"),
+                }
+            )
+        assistant = {"role": "assistant", "content": text, "tool_calls": tool_calls}
+        return {"assistant": assistant, "tool_calls": tool_calls, "text": text}
+
+    def _call_anthropic(self) -> dict[str, Any]:
+        response = self.client.messages.create(
+            model=self.config.model,
+            system=self.system_prompt,
+            max_tokens=2048,
+            messages=self._anthropic_messages(),
+            tools=self._anthropic_tools(),
+        )
+
+        text_parts: list[str] = []
+        tool_calls: list[dict[str, Any]] = []
+        content_blocks: list[dict[str, Any]] = []
+        for block in response.content:
+            if block.type == "text":
+                text_parts.append(block.text)
+                content_blocks.append({"type": "text", "text": block.text})
+            elif block.type == "tool_use":
+                payload = dict(block.input)
+                tool_calls.append({"id": block.id, "name": block.name, "input": payload})
+                content_blocks.append({"type": "tool_use", "id": block.id, "name": block.name, "input": payload})
+
+        assistant = {"role": "assistant", "content": content_blocks, "tool_calls": tool_calls}
+        return {"assistant": assistant, "tool_calls": tool_calls, "text": "\n".join(text_parts)}
+
+    def _run_tool(self, name: str, payload: dict[str, Any]) -> str:
+        tool = self.tools.get(name)
+        if not tool:
+            return f"Tool not found: {name}"
+        try:
+            return tool.execute(payload)
+        except Exception as exc:
+            return f"Tool error in {name}: {exc}"
+
+    def _openai_tools(self) -> list[dict[str, Any]]:
+        return [
+            {
+                "type": "function",
+                "function": {
+                    "name": tool.name,
+                    "description": tool.description,
+                    "parameters": tool.input_schema,
+                },
+            }
+            for tool in self.tools.values()
+        ]
+
+    def _anthropic_tools(self) -> list[dict[str, Any]]:
+        return [
+            {
+                "name": tool.name,
+                "description": tool.description,
+                "input_schema": tool.input_schema,
+            }
+            for tool in self.tools.values()
+        ]
+
+    def _openai_messages(self) -> list[dict[str, Any]]:
+        messages: list[dict[str, Any]] = []
+        if self.system_prompt:
+            messages.append({"role": "system", "content": self.system_prompt})
+
+        for item in self.history:
+            if item["role"] == "user":
+                messages.append({"role": "user", "content": item["content"]})
+            elif item["role"] == "assistant":
+                payload: dict[str, Any] = {"role": "assistant", "content": item.get("content", "")}
+                if item.get("tool_calls"):
+                    payload["tool_calls"] = [
+                        {
+                            "id": call["id"],
+                            "type": "function",
+                            "function": {
+                                "name": call["name"],
+                                "arguments": json.dumps(call["input"], ensure_ascii=False),
+                            },
+                        }
+                        for call in item["tool_calls"]
+                    ]
+                messages.append(payload)
+            elif item["role"] == "tool":
+                messages.append(
+                    {
+                        "role": "tool",
+                        "tool_call_id": item["tool_call_id"],
+                        "content": item["content"],
+                    }
+                )
+        return messages
+
+    def _anthropic_messages(self) -> list[dict[str, Any]]:
+        messages: list[dict[str, Any]] = []
+        for item in self.history:
+            if item["role"] == "user":
+                messages.append({"role": "user", "content": item["content"]})
+            elif item["role"] == "assistant":
+                content = item.get("content", "")
+                messages.append({"role": "assistant", "content": content if isinstance(content, list) else content or ""})
+            elif item["role"] == "tool":
+                messages.append(
+                    {
+                        "role": "user",
+                        "content": [
+                            {
+                                "type": "tool_result",
+                                "tool_use_id": item["tool_call_id"],
+                                "content": item["content"],
+                            }
+                        ],
+                    }
+                )
+        return messages
+
+
+def _load_file_config(path: Path) -> dict[str, Any]:
+    if not path.exists():
+        return {}
+    data = tomllib.loads(path.read_text(encoding="utf-8"))
+    if "cc_slim" in data and isinstance(data["cc_slim"], dict):
+        return dict(data["cc_slim"])
+    return {k: v for k, v in data.items() if not isinstance(v, dict)}
+
+
+def _pick(*values: Any) -> Any:
+    for value in values:
+        if value is None:
+            continue
+        if isinstance(value, str) and not value.strip():
+            continue
+        return value
+    return None
@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Optional
+
+import typer
+from rich.console import Console
+
+from cc_slim.engine import Agent, resolve_config
+from cc_slim.tools import build_default_tools
+
+app = typer.Typer(add_completion=False, no_args_is_help=False)
+console = Console()
+
+
+@app.command()
+def run(
+    prompt: Optional[str] = typer.Argument(None, help="单次执行的用户输入"),
+    provider: Optional[str] = typer.Option(None, help="模型提供方：openai 或 anthropic"),
+    model: Optional[str] = typer.Option(None, help="模型名称"),
+    api_key: Optional[str] = typer.Option(None, help="API Key，优先级最高"),
+    base_url: Optional[str] = typer.Option(None, help="可选的 API Base URL"),
+    cwd: Path = typer.Option(Path("."), help="工作区根目录"),
+    max_turns: Optional[int] = typer.Option(None, help="最大工具循环轮数"),
+) -> None:
+    root = cwd.resolve()
+    config = resolve_config(
+        root,
+        {
+            "provider": provider,
+            "model": model,
+            "api_key": api_key,
+            "base_url": base_url,
+            "max_turns": max_turns,
+        },
+    )
+    agent = Agent(config=config, tools=build_default_tools(root), workspace=root)
+
+    if prompt:
+        console.print(agent.reply(prompt))
+        return
+
+    console.print("[bold cyan]cc-slim[/bold cyan] REPL，输入 exit 或 quit 退出。")
+    while True:
+        try:
+            user_input = typer.prompt(">")
+        except (EOFError, KeyboardInterrupt):
+            console.print()
+            break
+
+        if user_input.strip().lower() in {"exit", "quit"}:
+            break
+        if not user_input.strip():
+            continue
+
+        try:
+            console.print(agent.reply(user_input))
+        except Exception as exc:  # pragma: no cover
+            console.print(f"[red]error:[/red] {exc}")
+
+
+if __name__ == "__main__":
+    app()
@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import json
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable
+
+
+@dataclass
+class Tool:
+    name: str
+    description: str
+    input_schema: dict[str, Any]
+    execute: Callable[[dict[str, Any]], str]
+
+
+def build_default_tools(workspace: Path) -> list[Tool]:
+    return [
+        Tool(
+            name="Read",
+            description="读取工作区内的文本文件。",
+            input_schema={
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "相对工作区的文件路径"},
+                },
+                "required": ["path"],
+                "additionalProperties": False,
+            },
+            execute=lambda data: read_tool(workspace, data),
+        ),
+        Tool(
+            name="Glob",
+            description="按 glob 模式查找工作区路径。",
+            input_schema={
+                "type": "object",
+                "properties": {
+                    "pattern": {"type": "string", "description": "例如 src/**/*.py 的 glob 模式"},
+                },
+                "required": ["pattern"],
+                "additionalProperties": False,
+            },
+            execute=lambda data: glob_tool(workspace, data),
+        ),
+        Tool(
+            name="Bash",
+            description="在工作区中执行一条 shell 命令。",
+            input_schema={
+                "type": "object",
+                "properties": {
+                    "command": {"type": "string", "description": "要执行的命令"},
+                },
+                "required": ["command"],
+                "additionalProperties": False,
+            },
+            execute=lambda data: bash_tool(workspace, data),
+        ),
+    ]
+
+
+def read_tool(workspace: Path, data: dict[str, Any]) -> str:
+    path = _safe_path(workspace, str(data["path"]))
+    if not path.exists():
+        return f"文件不存在: {path.relative_to(workspace)}"
+    if path.is_dir():
+        return f"目标是目录，不是文件: {path.relative_to(workspace)}"
+
+    text = path.read_text(encoding="utf-8", errors="replace")
+    lines = text.splitlines()
+    rendered = "\n".join(f"{i}: {line}" for i, line in enumerate(lines, start=1))
+    if len(rendered) > 12000:
+        rendered = rendered[:12000] + "\n...<truncated>"
+    return rendered or "(empty file)"
+
+
+def glob_tool(workspace: Path, data: dict[str, Any]) -> str:
+    pattern = str(data["pattern"])
+    matches = sorted({path.relative_to(workspace).as_posix() for path in workspace.glob(pattern) if path != workspace})
+    if not matches:
+        return "(no matches)"
+    return "\n".join(matches[:200])
+
+
+def bash_tool(workspace: Path, data: dict[str, Any]) -> str:
+    command = str(data["command"])
+    proc = subprocess.run(
+        command,
+        cwd=workspace,
+        shell=True,
+        capture_output=True,
+        text=True,
+        timeout=60,
+    )
+    payload = {
+        "command": command,
+        "returncode": proc.returncode,
+        "stdout": proc.stdout[-8000:],
+        "stderr": proc.stderr[-4000:],
+    }
+    return json.dumps(payload, ensure_ascii=False, indent=2)
+
+
+def _safe_path(workspace: Path, raw: str) -> Path:
+    root = workspace.resolve()
+    path = (root / raw).resolve()
+    if root not in path.parents and path != root:
+        raise ValueError("路径越过工作区边界")
+    return path