เจาะลึกการสร้าง Agentic Workflow ด้วย Kimi CLI: จากการแก้บั๊กอัตโนมัติสู่ระบบ Session Memory แบบไร้รอยต่อ

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 — การสำรวจแบบอ่านอย่างเดียว สร้างแผนการดำเนินงาน:
kimi --quiet --plan -w /content/demo_project -p "Plan adding SQLite persistence"
# เลือกโมเดลที่แตกต่างกันในขณะรัน (ต้องมีอยู่ใน config.toml):
kimi --quiet -m kimi-k2 -p "hello"
# Thinking mode (หากโมเดลรองรับ):
kimi --quiet --thinking -p "Prove sqrt(2) is irrational"
# Ralph loop — ส่งพรอมต์เดิมซ้ำๆ สำหรับงานใหญ่หนึ่งงาน
# จนกว่าเอเจนต์จะส่งเอาต์พุต <choice>STOP</choice> หรือครบกำหนดขีดจำกัด:
kimi --print --yolo --max-ralph-iterations 5 -w /content/demo_project \
-p "Keep improving test coverage; STOP when everything is covered."
# MCP tools — ให้ความสามารถเพิ่มเติมแก่ Kimi ผ่านไฟล์กำหนดค่า MCP:
# /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 ..."
# ส่งออกเซสชัน (context.jsonl, wire.jsonl, state.json) สำหรับการดีบัก:
kimi export --yes
# Web UI (ต้องมี tunnel บน Colab เช่น cloudflared/ngrok เนื่องจาก
# เป็นการเชื่อมต่อภายในเครื่อง): 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.")