Agentic AI: Tool Use, Planner Loops, and Production Agent Architecture

1interface AgentArchitecture {
2  // 1. LLM — the "brain": decides, plans, reasons
3  brain: LLMClient;
4 
5  // 2. Tools — the "hands": tools for interacting with the world
6  tools: Map<string, Tool>;
7 
8  // 3. Memory — the "memory": short- and long-term information storage
9  memory: {
10    shortTerm: ConversationHistory;  // current task context
11    longTerm: VectorMemory;          // persistent knowledge store
12  };
13 
14  // 4. Planning — the "coordinator": which tool, when, in what order
15  planner: PlanningStrategy;
16}
17 
18interface Tool {
19  name: string;
20  description: string;  // the LLM reads this description to select the tool
21  parameters: JSONSchema;
22  execute: (params: unknown) => Promise<ToolResult>;
23}

1import openai
2import json
3 
4client = openai.OpenAI()
5 
6tools = [
7    {
8        "type": "function",
9        "function": {
10            "name": "search_database",
11            "description": "Search the product database. Returns price, stock, and product details.",
12            "parameters": {
13                "type": "object",
14                "properties": {
15                    "query": {
16                        "type": "string",
17                        "description": "Search query"
18                    },
19                    "category": {
20                        "type": "string",
21                        "enum": ["elektronik", "giyim", "gıda"],
22                        "description": "Product category (optional)"
23                    },
24                    "max_price": {
25                        "type": "number",
26                        "description": "Maximum price (TRY)"
27                    }
28                },
29                "required": ["query"]
30            }
31        }
32    }
33]
34 
35response = client.chat.completions.create(
36    model="gpt-4.7",
37    messages=[
38        {"role": "user", "content": "Show me the best wireless headphones under 1000 TRY"}
39    ],
40    tools=tools,
41    tool_choice="auto"
42)
43 
44message = response.choices[0].message
45if message.tool_calls:
46    for call in message.tool_calls:
47        fn_name = call.function.name
48        fn_args = json.loads(call.function.arguments)
49        result = execute_tool(fn_name, fn_args)

1import anthropic
2 
3client = anthropic.Anthropic()
4 
5tools = [
6    {
7        "name": "get_weather",
8        "description": "Get current weather information for the specified city",
9        "input_schema": {
10            "type": "object",
11            "properties": {
12                "city": {
13                    "type": "string",
14                    "description": "City name"
15                },
16                "unit": {
17                    "type": "string",
18                    "enum": ["celsius", "fahrenheit"],
19                    "default": "celsius"
20                }
21            },
22            "required": ["city"]
23        }
24    }
25]
26 
27response = client.messages.create(
28    model="claude-opus-4-7",
29    max_tokens=1024,
30    tools=tools,
31    messages=[
32        {"role": "user", "content": "What's the weather like in Istanbul today?"}
33    ]
34)
35 
36for block in response.content:
37    if block.type == "tool_use":
38        result = execute_tool(block.name, block.input)
39        # Send the result back to the model
40        continue_response = client.messages.create(
41            model="claude-opus-4-7",
42            max_tokens=1024,
43            tools=tools,
44            messages=[
45                {"role": "user", "content": "What's the weather like in Istanbul today?"},
46                {"role": "assistant", "content": response.content},
47                {"role": "user", "content": [
48                    {"type": "tool_result", "tool_use_id": block.id, "content": str(result)}
49                ]}
50            ]
51        )

1async function executeParallelTools(
2  toolCalls: ToolCall[],
3  tools: Map<string, Tool>
4): Promise<ToolResult[]> {
5  // Run in parallel — reduces latency
6  return Promise.all(
7    toolCalls.map(async (call) => {
8      const tool = tools.get(call.name);
9      if (!tool) throw new Error(`Tool not found: ${call.name}`);
10 
11      try {
12        const result = await tool.execute(call.params);
13        return { id: call.id, success: true, result };
14      } catch (error) {
15        return { id: call.id, success: false, error: String(error) };
16      }
17    })
18  );
19}

1class PlanAndExecuteAgent:
2    def __init__(self, llm, tools):
3        self.llm = llm
4        self.tools = tools
5 
6    async def run(self, task: str) -> str:
7        # 1. Task analysis and plan generation
8        plan = await self.create_plan(task)
9 
10        results = []
11        for step in plan['steps']:
12            result = await self.execute_step(step, results)
13            results.append(result)
14 
15            # Update the plan if needed
16            if result.get('needs_replan'):
17                plan = await self.replan(task, plan, results)
18 
19        return await self.synthesize(task, results)
20 
21    async def create_plan(self, task: str) -> dict:
22        prompt = f"""Break the task into sub-steps. Each step should be independently executable.
23 
24TASK: {task}
25AVAILABLE TOOLS: {list(self.tools.keys())}
26 
27Return JSON:
28{{"steps": [{{"id": 1, "description": "...", "tool": "...", "depends_on": []}}]}}"""
29        response = await self.llm.complete(prompt)
30        return json.loads(response)

1class ReflexionAgent:
2    def __init__(self, llm, tools, max_trials: int = 3):
3        self.llm = llm
4        self.tools = tools
5        self.max_trials = max_trials
6 
7    async def run(self, task: str) -> str:
8        reflections: list[str] = []
9 
10        for trial in range(self.max_trials):
11            result = await self.attempt(task, reflections)
12 
13            if result['success']:
14                return result['output']
15 
16            reflection = await self.reflect(task, result)
17            reflections.append(reflection)
18 
19        raise RuntimeError(f"Task could not be solved in {self.max_trials} attempts")
20 
21    async def reflect(self, task: str, result: dict) -> str:
22        prompt = f"""The task failed. What did you do wrong?
23 
24TASK: {task}
25ERROR RECEIVED: {result.get('error')}
26ACTIONS TAKEN: {result.get('actions')}
27 
28Write a short, specific analysis (max 100 words):"""
29        return await self.llm.complete(prompt)

1class ConversationMemory {
2  private messages: Message[] = [];
3  private readonly maxTokens: number;
4 
5  constructor(maxTokens = 8000) {
6    this.maxTokens = maxTokens;
7  }
8 
9  add(message: Message): void {
10    this.messages.push(message);
11    this.enforceLimit();
12  }
13 
14  private async enforceLimit(): Promise<void> {
15    const currentTokens = this.estimateTokenCount();
16 
17    if (currentTokens > this.maxTokens * 0.8) {
18      // Summarize old messages — keep the last 4
19      const toSummarize = this.messages.slice(0, -4);
20      const summary = await this.summarize(toSummarize);
21 
22      this.messages = [
23        { role: 'system', content: `Previous conversation summary: ${summary}` },
24        ...this.messages.slice(-4)
25      ];
26    }
27  }
28 
29  private estimateTokenCount(): number {
30    return this.messages.reduce((total, m) => total + m.content.length / 4, 0);
31  }
32}

1class SemanticMemory:
2    def __init__(self, vector_db, embedder):
3        self.db = vector_db
4        self.embedder = embedder
5 
6    async def store(self, key: str, content: str, metadata: dict = {}):
7        embedding = await self.embedder.embed(content)
8        await self.db.upsert({
9            "id": key,
10            "vector": embedding,
11            "metadata": {"content": content, **metadata}
12        })
13 
14    async def recall(self, query: str, k: int = 5) -> list[str]:
15        query_vec = await self.embedder.embed(query)
16        results = await self.db.search(query_vec, k)
17        return [r.metadata["content"] for r in results]
18 
19    async def build_context_block(self, query: str) -> str:
20        memories = await self.recall(query)
21        if not memories:
22            return ""
23        lines = "
24".join(f"- {m}" for m in memories)
25        return f"
26 
27Relevant historical info:
28{lines}"

1import autogen
2 
3config_list = [{"model": "gpt-4.7", "api_key": "YOUR_KEY"}]
4 
5planner = autogen.AssistantAgent(
6    name="Planner",
7    system_message="Break tasks into sub-steps and coordinate them.",
8    llm_config={"config_list": config_list}
9)
10 
11coder = autogen.AssistantAgent(
12    name="Coder",
13    system_message="Write and execute Python code.",
14    llm_config={"config_list": config_list}
15)
16 
17critic = autogen.AssistantAgent(
18    name="Critic",
19    system_message="Review the code, suggest fixes and improvements.",
20    llm_config={"config_list": config_list}
21)
22 
23user_proxy = autogen.UserProxyAgent(
24    name="UserProxy",
25    human_input_mode="NEVER",
26    code_execution_config={"work_dir": "coding"}
27)
28 
29group_chat = autogen.GroupChat(
30    agents=[planner, coder, critic, user_proxy],
31    messages=[],
32    max_round=10
33)
34manager = autogen.GroupChatManager(groupchat=group_chat, llm_config={"config_list": config_list})

1class MultiAgentOrchestrator {
2  private workers = new Map<string, WorkerAgent>();
3 
4  register(role: string, agent: WorkerAgent): void {
5    this.workers.set(role, agent);
6  }
7 
8  async executeTask(task: ComplexTask): Promise<TaskResult> {
9    const subTasks = await this.decompose(task);
10 
11    // Run independent tasks in parallel
12    const results = new Map<string, SubTaskResult>();
13    const groups = this.groupByDependency(subTasks);
14 
15    for (const group of groups) {
16      const groupResults = await Promise.all(
17        group.map((st) => {
18          const worker = this.workers.get(st.requiredRole);
19          if (!worker) throw new Error(`No worker for role: ${st.requiredRole}`);
20          return worker.execute(st, results);
21        })
22      );
23      groupResults.forEach((r, i) => results.set(group[i].id, r));
24    }
25 
26    return this.synthesize(task, results);
27  }
28}

1const enum AgentErrorType {
2  ToolTimeout = 'TOOL_TIMEOUT',
3  ToolRateLimit = 'TOOL_RATE_LIMIT',
4  ContextOverflow = 'LLM_CONTEXT_OVERFLOW',
5  ToolExecutionError = 'TOOL_EXECUTION_ERROR',
6  PlanningFailure = 'PLANNING_FAILURE',
7  MaxIterations = 'MAX_ITERATIONS_EXCEEDED',
8}
9 
10interface RetryStrategy {
11  retry: boolean;
12  maxRetries?: number;
13  backoffMs?: number;
14  exponential?: boolean;
15  action?: 'COMPRESS_CONTEXT' | 'REPLAN';
16  escalate?: boolean;
17}
18 
19const ERROR_STRATEGIES: Record<AgentErrorType, RetryStrategy> = {
20  [AgentErrorType.ToolTimeout]: { retry: true, maxRetries: 3, backoffMs: 1000, exponential: true },
21  [AgentErrorType.ToolRateLimit]: { retry: true, maxRetries: 5, backoffMs: 5000, exponential: true },
22  [AgentErrorType.ContextOverflow]: { retry: true, maxRetries: 1, action: 'COMPRESS_CONTEXT', backoffMs: 0 },
23  [AgentErrorType.PlanningFailure]: { retry: true, maxRetries: 2, action: 'REPLAN', backoffMs: 0 },
24  [AgentErrorType.MaxIterations]: { retry: false, escalate: true },
25  [AgentErrorType.ToolExecutionError]: { retry: true, maxRetries: 2, backoffMs: 500, exponential: false },
26};

1import time
2 
3class CircuitBreaker:
4    def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 60):
5        self.failure_threshold = failure_threshold
6        self.recovery_timeout = recovery_timeout
7        self.failure_count = 0
8        self.last_failure_time: float | None = None
9        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
10 
11    async def call(self, func, *args, **kwargs):
12        if self.state == "OPEN":
13            elapsed = time.time() - (self.last_failure_time or 0)
14            if elapsed > self.recovery_timeout:
15                self.state = "HALF_OPEN"
16            else:
17                raise RuntimeError("Circuit breaker OPEN — tool temporarily disabled")
18 
19        try:
20            result = await func(*args, **kwargs)
21            self._on_success()
22            return result
23        except Exception as exc:
24            self._on_failure()
25            raise
26 
27    def _on_success(self):
28        self.failure_count = 0
29        self.state = "CLOSED"
30 
31    def _on_failure(self):
32        self.failure_count += 1
33        self.last_failure_time = time.time()
34        if self.failure_count >= self.failure_threshold:
35            self.state = "OPEN"

1import structlog
2import time
3 
4logger = structlog.get_logger()
5 
6class ObservableAgent:
7    async def execute_step(self, step: dict) -> dict:
8        step_id = step['id']
9        step_type = step['type']
10 
11        logger.info("agent_step_start", step_id=step_id, step_type=step_type)
12        start_time = time.time()
13 
14        try:
15            result = await self._run_step(step)
16            duration_ms = (time.time() - start_time) * 1000
17 
18            logger.info(
19                "agent_step_complete",
20                step_id=step_id,
21                duration_ms=round(duration_ms, 1),
22                success=True
23            )
24            return result
25 
26        except Exception as exc:
27            logger.error(
28                "agent_step_failed",
29                step_id=step_id,
30                error=str(exc),
31                duration_ms=round((time.time() - start_time) * 1000, 1)
32            )
33            raise

1const customerSupportAgent = {
2  model: "claude-opus-4-7",
3  maxIterations: 8,
4 
5  tools: [
6    {
7      name: "get_order_status",
8      description: "Returns the customer's order status and shipping information",
9      parameters: {
10        order_id: { type: "string", required: true },
11        customer_email: { type: "string", required: false }
12      }
13    },
14    {
15      name: "initiate_return",
16      description: "Initiates a return request and returns return instructions",
17      parameters: {
18        order_id: { type: "string", required: true },
19        reason: {
20          type: "string",
21          enum: ["defective", "wrong_item", "not_as_described", "change_of_mind"]
22        }
23      }
24    },
25    {
26      name: "search_product_info",
27      description: "Product features, stock status, and price information",
28      parameters: {
29        query: { type: "string", required: true }
30      }
31    },
32    {
33      name: "escalate_to_human",
34      description: "Escalates complex or sensitive situations to a human agent",
35      parameters: {
36        reason: { type: "string" },
37        priority: { type: "string", enum: ["low", "medium", "high", "urgent"] },
38        summary: { type: "string" }
39      }
40    }
41  ],
42 
43  systemPrompt: `You are [Company]'s customer support assistant.
44Be polite, solution-oriented, and efficient.
45Don't perform sensitive actions without verifying customer information.
46Escalate to a human agent if you can't find a solution.`
47};