بناء مسارات عمل ترميز وكيل غير تفاعلية باستخدام Kimi CLI وتدفق JSONL والاختبار وذاكرة الجلسة من Moonshot AI

import os, subprocess, textwrap, json, getpass, pathlib, shutil
HOME = pathlib.Path.home()
def sh(cmd, check=True, env=None, cwd=None):
   """Run a shell command, stream its output, return CompletedProcess."""
   print(f"\n$ {cmd}")
   e = {**os.environ, **(env or {})}
   r = subprocess.run(cmd, shell=True, env=e, cwd=cwd,
                      capture_output=True, text=True)
   if r.stdout: print(r.stdout)
   if r.stderr: print(r.stderr[-2000:])
   if check and r.returncode != 0:
       raise RuntimeError(f"Command failed ({r.returncode}): {cmd}")
   return r
print("=" * 70, "\nPART 1: Installing uv + Kimi CLI\n", "=" * 70)
sh("curl -LsSf https://astral.sh/uv/install.sh | sh")
UV_BIN = str(HOME / ".local" / "bin")
os.environ["PATH"] = f"{UV_BIN}:{os.environ['PATH']}"
sh("uv tool install --python 3.13 kimi-cli")
sh("kimi --version")

print("=" * 70, "\nPART 2: Configuring API access\n", "=" * 70)
try:
   from google.colab import userdata
   API_KEY = userdata.get("MOONSHOT_API_KEY")
   print("Loaded key from Colab Secrets.")
except Exception:
   API_KEY = getpass.getpass("Paste your Moonshot API key (hidden): ")
BASE_URL   = "https://api.moonshot.ai/v1"
MODEL_NAME = "kimi-k2-0711-preview"
kimi_dir = HOME / ".kimi"
kimi_dir.mkdir(exist_ok=True)
(kimi_dir / "config.toml").write_text(textwrap.dedent(f"""\
   default_model = "kimi-k2"
   [providers.moonshot]
   type = "kimi"
   base_url = "{BASE_URL}"
   api_key = "{API_KEY}"
   [models.kimi-k2]
   provider = "moonshot"
   model = "{MODEL_NAME}"
   max_context_size = 131072
"""))
print("Wrote ~/.kimi/config.toml")

def kimi(prompt, work_dir=".", yolo=False, cont=False, quiet=True,
        stream_json=False, extra="", timeout=600):
   """Run one headless Kimi CLI turn and return its stdout."""
   flags = []
   if stream_json:
       flags.append("--print --output-format stream-json")
   elif quiet:
       flags.append("--quiet")
   else:
       flags.append("--print")
   if yolo: flags.append("--yolo")
   if cont: flags.append("--continue")
   flags.append(f'-w "{work_dir}"')
   if extra: flags.append(extra)
   cmd = f'kimi {" ".join(flags)} -p "{prompt}"'
   print(f"\n$ {cmd}\n" + "-" * 60)
   r = subprocess.run(cmd, shell=True, capture_output=True,
                      text=True, timeout=timeout)
   out = r.stdout.strip()
   print(out if out else r.stderr[-1500:])
   return out

print("=" * 70, "\nPART 4: Demo A — codebase Q&A\n", "=" * 70)
proj = pathlib.Path("/content/demo_project")
if proj.exists(): shutil.rmtree(proj)
(proj / "app").mkdir(parents=True)
(proj / "app" / "inventory.py").write_text(textwrap.dedent("""\
   class Inventory:
       def __init__(self):
           self.items = {}
       def add(self, name, qty):
           self.items[name] = self.items.get(name, 0) + qty
       def remove(self, name, qty):
           # BUG: allows negative stock and KeyError on missing items
           self.items[name] = self.items[name] - qty
       def total(self):
           return sum(self.items.values())
"""))
(proj / "app" / "main.py").write_text(textwrap.dedent("""\
   from inventory import Inventory
   inv = Inventory()
   inv.add("widget", 10)
   inv.remove("widget", 3)
   print("Total stock:", inv.total())
"""))
(proj / "README.md").write_text("# Demo: a tiny inventory service\n")
kimi("Summarize this project's structure and purpose in under 120 words, "
    "then list any bugs or design risks you can spot in inventory.py.",
    work_dir=str(proj))

print("=" * 70, "\nPART 5: Demo B — Kimi fixes the bug & adds tests\n", "=" * 70)
kimi("Fix the bugs in app/inventory.py: remove() must raise KeyError->ValueError "
    "for unknown items and never allow negative stock. Then create tests.py at "
    "the project root using unittest covering add/remove/total and edge cases, "
    "run it with 'python -m unittest tests -v', and iterate until all tests pass. "
    "Finally print the test results.",
    work_dir=str(proj), yolo=True, extra="--max-steps-per-turn 30")
print("\n--- Files after Kimi's edits ---")
for f in sorted(proj.rglob("*.py")):
   print(f"\n### {f.relative_to(proj)} ###\n{f.read_text()}")
sh("python -m unittest tests -v", cwd=str(proj), check=False)

print("=" * 70, "\nPART 6: Demo C — machine-readable JSONL events\n", "=" * 70)
raw = kimi("In one sentence, what does app/main.py print when run?",
          work_dir=str(proj), stream_json=True, quiet=False)
print("\nParsed event types:")
for line in raw.splitlines():
   try:
       evt = json.loads(line)
       print(" •", evt.get("type", "?"), "-",
             str(evt)[:100].replace("\n", " "))
   except json.JSONDecodeError:
       pass
print("=" * 70, "\nPART 7: Demo D — conversational memory\n", "=" * 70)
kimi("Remember this: our release codename is BLUE-FALCON.", work_dir=str(proj))
kimi("What is our release codename? Answer with just the codename.",
    work_dir=str(proj), cont=True)
print("=" * 70, "\nPART 8: Power-user reference\n", "=" * 70)
print(textwrap.dedent("""
 # Plan mode — read-only exploration, produces an implementation plan:
 kimi --quiet --plan -w /content/demo_project -p "Plan adding SQLite persistence"
 # Pick a different model at runtime (must exist in config.toml):
 kimi --quiet -m kimi-k2 -p "hello"
 # Thinking mode (if the model supports it):
 kimi --quiet --thinking -p "Prove sqrt(2) is irrational"
 # Ralph loop — feed the same prompt repeatedly for one big task
 # until the agent outputs <choice>STOP</choice> or the limit hits:
 kimi --print --yolo --max-ralph-iterations 5 -w /content/demo_project \\
      -p "Keep improving test coverage; STOP when everything is covered."
 # MCP tools — give Kimi extra capabilities via an MCP config file:
 #   /content/mcp.json -> {"mcpServers": {"context7":
 #        {"url": "https://mcp.context7.com/mcp",
 #         "headers": {"CONTEXT7_API_KEY": "YOUR_KEY"}}}}
 kimi --quiet --mcp-config-file /content/mcp.json -p "Use context7 to ..."
 # Export a session (context.jsonl, wire.jsonl, state.json) for debugging:
 kimi export --yes
 # Web UI (needs a tunnel on Colab, e.g. cloudflared/ngrok, since it
 # binds locally): kimi web --no-open --port 5494
"""))
print("Tutorial complete ✔ — Kimi CLI is installed, authenticated, and has "
     "explored, fixed, tested, and remembered a real project headlessly.")