Arsitektur Agent

Diagram arsitektur Kai

Sebelum lo bisa apply 10 pilar SOUL, lo perlu arsitektur kode yang ngedukung. Ini blueprint minimal yang gua pake di Kai.

Komponen utama

agent/
├── main.py                 # Entry point: Telegram handler, command router
├── soul.md                 # File konstitusi (dibaca tiap call)
├── data/
│   ├── memory.json         # User preferences, koreksi, fakta stabil
│   └── history/
│       └── <user_id>.json  # Per-user chat history
├── credentials/
│   ├── github.env          # chmod 600
│   ├── wallet.env          # chmod 600
│   └── README.txt
└── .env                    # TELEGRAM_BOT_TOKEN, OPENAI_API_KEY

Flow: dari pesan user sampai respon

1. User kirim pesan via Telegram
   ↓
2. Handler terima, load history user
   ↓
3. Build system prompt:
   - Baca SOUL.md
   - Append memory notes
   - Append credential file list (nama doang, BUKAN isi)
   - Append konteks ekstra (tanggal, dll)
   ↓
4. LLM call dengan: system_prompt + history + new_message
   ↓
5. LLM respon: bisa berisi <tool_use>...</tool_use> atau plain text
   ↓
6. Parse tool_use:
   - Kalau ada → cek risk field
     - low/medium → auto-execute, hasil dikirim balik ke LLM
     - high → simpen ke pending_commands, minta konfirmasi user
   - Kalau ga ada tool_use → reply final ke user
   ↓
7. Multi-step:
   - Setelah tool_use dieksekusi → loop ke step 4 (dengan history yang udah ditambah tool result)
   - Max iterasi (misalnya 12 step) biar ga infinite loop
   ↓
8. Save history (dengan secret redaction)

Komponen 1: Build system prompt

def build_system_prompt():
    memory = load_memory()
    soul = load_soul()  # Read SOUL.md from disk every time
    creds = list_credentials()  # Cuma nama file
    today = datetime.now().strftime("%A, %d %B %Y %H:%M")

    memory_text = ""
    if memory.get("notes"):
        memory_text = "\n\nYang lo ingat tentang user:\n" + \
            "\n".join(f"- {n}" for n in memory["notes"][-20:])

    creds_text = ""
    if creds:
        creds_text = "\n\nCredential available:\n" + \
            "\n".join(f"- {f}" for f in creds)

    return f"{soul}\n\nHari ini: {today}{creds_text}{memory_text}"

Kuncinya: load_soul() baca file tiap call. Jadi lo bisa edit SOUL.md kapan aja tanpa restart bot.

Komponen 2: Tool use parser

Gua pakai format <tool_use>{...}</tool_use> dengan JSON di dalamnya:

import re, json

def extract_tool_calls(text: str) -> list:
    results = []
    pattern = re.compile(r'<tool_use>(.*?)</tool_use>', re.DOTALL)
    for m in pattern.finditer(text):
        try:
            results.append(json.loads(m.group(1).strip()))
        except json.JSONDecodeError:
            pass
    return results

Format yang model harus output:

<tool_use>{"name":"run_in_terminal","arguments":{"command":"ls -la","risk":"low","explanation":"cek isi folder"}}</tool_use>

Detail di Format Tool Use.

Komponen 3: Risk-gating

def execute_with_gating(call, chat_id, context):
    cmd = call["arguments"]["command"]
    risk = call["arguments"].get("risk", "high")  # default high if missing

    if risk in ("low", "medium"):
        return run_command(cmd)  # auto-execute
    elif risk == "high":
        pending_commands[chat_id] = cmd
        return None  # signal: minta konfirmasi user

Detail di Autonomy & Risk-Gating.

Komponen 4: Agentic loop

def ask_ai_agentic(chat_id, user_text, max_steps=12):
    history = load_history(chat_id)
    messages = [
        {"role": "system", "content": build_system_prompt()},
        *history,
        {"role": "user", "content": user_text}
    ]

    for step in range(max_steps):
        response = llm.chat(messages=messages)
        answer = response.content

        calls = extract_tool_calls(answer)
        if not calls:
            # Final answer
            history.append({"role": "assistant", "content": answer})
            save_history(chat_id, history)
            return answer

        # Execute tools
        messages.append({"role": "assistant", "content": answer})
        for call in calls:
            result = execute_with_gating(call, chat_id)
            if result is None:
                return f"Mau jalanin command high-risk: {call['arguments']['command']}\nBalas 'ya' untuk lanjut."
            messages.append({"role": "user", "content": f"Tool result: {result}"})

    return "Max steps tercapai. Belum kelar."

Komponen 5: Secret redaction

History harus sanitize sebelum di-save, biar credential ga ke-log:

SECRET_PATTERNS = [
    (re.compile(r'ghp_[A-Za-z0-9]{36}'), '[REDACTED_GITHUB_PAT]'),
    (re.compile(r'0x[a-fA-F0-9]{64}'), '[REDACTED_PRIVATE_KEY]'),
    (re.compile(r'AKIA[0-9A-Z]{16}'), '[REDACTED_AWS_KEY]'),
    (re.compile(r'sk-[A-Za-z0-9]{40,}'), '[REDACTED_OPENAI_KEY]'),
    (re.compile(r'xox[bp]-[0-9-A-Za-z]+'), '[REDACTED_SLACK_TOKEN]'),
    (re.compile(r'-----BEGIN [A-Z ]+PRIVATE KEY-----.*?-----END [A-Z ]+PRIVATE KEY-----', re.DOTALL), '[REDACTED_PEM]'),
]

def redact_secrets(text: str) -> str:
    for pattern, replacement in SECRET_PATTERNS:
        text = pattern.sub(replacement, text)
    return text

Apply ini di save_history() sebelum write ke disk.

Komponen 6: Per-user lock & cancel

Biar user ga bisa spam pesan yang bikin race condition:

chat_locks = {}  # chat_id -> asyncio.Lock
chat_busy = {}   # chat_id -> {"desc": ..., "start_time": ...}
chat_cancel = {} # chat_id -> bool

async def handle_message(update, context):
    chat_id = update.effective_chat.id
    lock = chat_locks.setdefault(chat_id, asyncio.Lock())

    # Fast path: kalo lagi busy + user nanya status
    if chat_busy.get(chat_id) and is_status_query(update.message.text):
        busy = chat_busy[chat_id]
        elapsed = int(time.time() - busy["start_time"])
        await update.message.reply_text(f"Lagi proses: {busy['desc']} ({elapsed}s)")
        return

    async with lock:
        chat_busy[chat_id] = {"desc": update.message.text[:40], "start_time": time.time()}
        try:
            result = ask_ai_agentic(chat_id, update.message.text)
            await update.message.reply_text(result)
        finally:
            chat_busy.pop(chat_id, None)

Komponen 7: Konfirmasi flow

Kalau ada pending_command (high-risk), pesan user berikutnya di-check apakah "ya" atau bukan:

async def handle_message(update, context):
    chat_id = update.effective_chat.id
    text = update.message.text.strip().lower()

    if chat_id in pending_commands:
        if text in ("ya", "y", "yes", "ok", "go"):
            cmd = pending_commands.pop(chat_id)
            result = run_command(cmd)
            await update.message.reply_text(f"Done. Result: {result[:500]}")
            return
        else:
            pending_commands.pop(chat_id)
            await update.message.reply_text("Dibatalin.")
            return

    # ... normal flow

Pola yang harus diikuti

1. SOUL.md file di luar kode — biar bisa edit tanpa restart
2. Credential di folder terpisah — chmod 700/600, ga commit ke git
3. History per-user — file <user_id>.json, dengan secret redaction
4. Tool use format yang konsisten — <tool_use>{...}</tool_use> JSON
5. Risk field wajib di setiap tool_use — biar gating jalan
6. Per-user lock — biar serialize, ga race
7. Multi-step max iter — biar ga infinite loop
8. Fast path untuk status query — biar user ga di-block

Detail per komponen ada di bagian Praktik.