AI Agent Architecture — Tool Use, Memory, Planning, Multi-Agent & Guardrails

# =============================================
# Function Calling dengan OpenAI
# =============================================
from openai import OpenAI
import json

client = OpenAI()

# Definisikan tools
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Dapatkan cuaca saat ini di lokasi tertentu",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "Nama kota, contoh: Jakarta, Bandung"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Satuan suhu"
                    }
                },
                "required": ["location"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_database",
            "description": "Cari informasi dari database produk",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Kata kunci pencarian"},
                    "category": {"type": "string", "description": "Filter kategori"}
                },
                "required": ["query"]
            }
        }
    }
]

# Implementasi fungsi
def get_weather(location, unit="celsius"):
    # Di production, ini akan call weather API
    return {"location": location, "temp": 32, "condition": "Cerah", "unit": unit}

def search_database(query, category=None):
    return {"results": [{"name": "Produk A", "price": 150000}]}

# Agent conversation
messages = [
    {"role": "system", "content": "Anda adalah asisten yang bisa menggunakan tools."},
    {"role": "user", "content": "Bagaimana cuaca di Jakarta hari ini?"}
]

response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)

# Check if LLM wants to call a tool
msg = response.choices[0].message
if msg.tool_calls:
    for tool_call in msg.tool_calls:
        func_name = tool_call.function.name
        args = json.loads(tool_call.function.arguments)
        
        # Execute the function
        if func_name == "get_weather":
            result = get_weather(**args)
        
        # Add function result to messages
        messages.append(msg)
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(result)
        })
    
    # Get final response
    final = client.chat.completions.create(
        model="gpt-4o", messages=messages, tools=tools
    )
    print(final.choices[0].message.content)

# =============================================
# Agent Memory System
# =============================================
from collections import deque
from datetime import datetime

class AgentMemory:
    def __init__(self, max_short_term=20):
        # Short-term: conversation window
        self.short_term = deque(maxlen=max_short_term)
        # Long-term: persistent facts
        self.long_term = {}
        # Working: current task context
        self.working = {}
    
    def add_message(self, role, content):
        self.short_term.append({
            "role": role, "content": content,
            "timestamp": datetime.now().isoformat()
        })
    
    def add_fact(self, key, value):
        self.long_term[key] = {"value": value, "updated": datetime.now()}
    
    def get_context(self):
        return list(self.short_term)
    
    def get_fact(self, key):
        return self.long_term.get(key, {}).get("value")
    
    def search_facts(self, query):
        results = []
        for key, val in self.long_term.items():
            if query.lower() in key.lower() or query.lower() in str(val).lower():
                results.append((key, val["value"]))
        return results

# Usage
memory = AgentMemory()
memory.add_message("user", "Nama saya Budi")
memory.add_message("assistant", "Halo Budi, ada yang bisa saya bantu?")
memory.add_fact("user_name", "Budi")
memory.add_fact("user_preference", "suka Python")

print(memory.get_fact("user_name"))  # "Budi"

# =============================================
# Planning Strategies untuk Agent
# =============================================

# ----- 1. Plan-and-Execute -----
# Agent membuat rencana lengkap, lalu eksekusi satu per satu
PLAN_PROMPT = """Buat rencana langkah demi langkah untuk menyelesaikan tugas ini.
Setiap langkah harus spesifik dan bisa dieksekusi.

Tugas: {task}
Tools tersedia: {tools}

Format output:
1. [langkah 1]
2. [langkah 2]
...
"""

# ----- 2. ReAct (Reasoning + Acting) -----
# Pikir → Aksi → Observasi → Ulangi
REACT_PROMPT = """Jawab pertanyaan menggunakan format berikut:

Thought: [pikirkan apa yang perlu dilakukan]
Action: [nama_tool]
Action Input: [input untuk tool]
Observation: [hasil dari tool]
... (ulangi sampai cukup informasi)
Thought: [pikirkan apakah sudah cukup]
Final Answer: [jawaban akhir]
"""

# ----- 3. Reflexion (Self-improvement) -----
# Setelah eksekusi, agent merefleksi dan memperbaiki
REFLECTION_PROMPT = """Evaluasi hasil eksekusi berikut:
Tugas: {task}
Hasil: {result}
Error: {errors}

Apakah hasilnya memuaskan? Jika tidak, apa yang perlu diperbaiki?
"""

# =============================================
# Complete Agent Loop
# =============================================
from openai import OpenAI
import json

class AIAgent:
    def __init__(self, tools, system_prompt, max_iterations=10):
        self.client = OpenAI()
        self.tools = tools
        self.system_prompt = system_prompt
        self.max_iterations = max_iterations
        self.memory = []
    
    def run(self, user_input):
        messages = [
            {"role": "system", "content": self.system_prompt},
            *self.memory,
            {"role": "user", "content": user_input}
        ]
        
        for i in range(self.max_iterations):
            response = self.client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                tools=self.tools,
                tool_choice="auto"
            )
            
            msg = response.choices[0].message
            
            # If LLM wants to use a tool
            if msg.tool_calls:
                messages.append(msg)
                
                for tool_call in msg.tool_calls:
                    result = self._execute_tool(
                        tool_call.function.name,
                        json.loads(tool_call.function.arguments)
                    )
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": json.dumps(result)
                    })
                
                continue  # Loop again
            
            # Final answer
            answer = msg.content
            self.memory.append({"role": "user", "content": user_input})
            self.memory.append({"role": "assistant", "content": answer})
            return answer
        
        return "Mencapai batas iterasi maksimum."
    
    def _execute_tool(self, name, args):
        tool_map = {t["function"]["name"]: t for t in self.tools}
        # Execute based on tool name
        # ... implementasi masing-masing tool
        return {"status": "success", "result": "..."}

# Usage
agent = AIAgent(tools=tools, system_prompt="Anda adalah asisten AI...")
result = agent.run("Berapa harga emas hari ini dalam Rupiah?")
print(result)

# =============================================
# Multi-Agent dengan LangGraph
# =============================================
# pip install langgraph langchain-openai

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated

class AgentState(TypedDict):
    task: str
    research: str
    code: str
    final_output: str

def researcher(state):
    """Agent yang mencari informasi."""
    # Call LLM for research
    return {"research": "Hasil riset tentang topik..."}

def coder(state):
    """Agent yang menulis kode."""
    research = state["research"]
    return {"code": "import pandas as pd\n..."}

def reviewer(state):
    """Agent yang mereview hasil."""
    return {"final_output": "Review selesai, kode valid."}

# Build graph
graph = StateGraph(AgentState)
graph.add_node("researcher", researcher)
graph.add_node("coder", coder)
graph.add_node("reviewer", reviewer)

graph.set_entry_point("researcher")
graph.add_edge("researcher", "coder")
graph.add_edge("coder", "reviewer")
graph.add_edge("reviewer", END)

# Compile & run
app = graph.compile()
result = app.invoke({"task": "Buat dashboard penjualan"})

# =============================================
# Guardrails untuk AI Agent
# =============================================

class AgentGuardrails:
    def __init__(self):
        self.max_iterations = 10
        self.blocked_actions = ["delete_database", "send_money"]
        self.max_tool_calls = 20
        self.allowed_domains = ["api.example.com", "database.internal"]
    
    def check_input(self, user_input):
        """Filter input berbahaya."""
        blocked = ["ignore instructions", "system prompt", "jailbreak"]
        for b in blocked:
            if b in user_input.lower():
                return False, "Input ditolak: mengandung instruksi terlarang"
        return True, "OK"
    
    def check_tool_call(self, tool_name, args):
        """Validasi sebelum eksekusi tool."""
        if tool_name in self.blocked_actions:
            return False, f"Tool '{tool_name}' diblokir"
        if tool_name == "http_request":
            url = args.get("url", "")
            if not any(d in url for d in self.allowed_domains):
                return False, f"Domain tidak diizinkan: {url}"
        return True, "OK"
    
    def check_output(self, output):
        """Filter output berbahaya."""
        sensitive = ["password", "api_key", "secret"]
        for s in sensitive:
            if s in output.lower():
                return False, "Output mengandung informasi sensitif"
        return True, "OK"

# Usage dalam agent loop
guardrails = AgentGuardrails()
ok, msg = guardrails.check_input(user_input)
if not ok:
    print(f"Ditolak: {msg}")