agentic-ai

# Agentic AI — Implementation Guide Covers: **Agent Architectures · Tool Use · Memory Systems · Planning · Evaluation · Multi-Agent Systems** ----- ## Understanding Agentic AI ### What Makes an AI "Agentic"? An agentic AI system differs from traditional AI assistants in several fundamental ways. While a standard language model responds to each prompt independently, an agentic system maintains state across interactions, takes autonomous actions to achieve goals, and can plan multi-step sequences of operations. The key characteristics that define agentic AI include: autonomy in decision-making without requiring constant human guidance, the ability to plan and execute multi-step workflows, the capacity to use external tools and APIs to interact with the world, memory systems that preserve context across interactions, and feedback mechanisms that enable learning and adaptation. **Agentic systems can be categorized by their complexity:** Simple reflex agents respond to stimuli based on predetermined rules. Goal-based agents work toward specific objectives using planning algorithms. Utility-based agents maximize expected utility through optimization. Learning agents improve performance through experience. Multi-agent systems involve multiple AI agents collaborating or competing. **Common Agent Architectures:** - **ReAct (Reason + Act)** — Combines reasoning traces with action execution - **Reflexion** — Adds verbal reinforcement learning for self-reflection - **Tool Use Agents** — Integrate external tools and APIs into reasoning - **Plan-and-Execute** — Separates planning from execution phases - **Multi-Agent** — Multiple specialized agents working together ### When to Use Agentic Systems Agentic systems excel in scenarios requiring complex multi-step reasoning, dynamic tool orchestration, persistent context across sessions, autonomous decision-making, iterative refinement, or coordination of multiple specialized components. They are particularly valuable for building AI assistants that can take actions rather than just generating text. Traditional completion-based AI remains superior for simple question-answering, content generation, summarization, and single-turn interactions. The complexity of agentic systems is justified when the task genuinely requires persistent state, tool use, or multi-step execution. ----- ## Agent Architecture Design ### Core Agent Loop ```python from dataclasses import dataclass, field from typing import List, Dict, Any, Optional, Callable from enum import Enum from datetime import datetime class AgentState(Enum): IDLE = "idle" REASONING = "reasoning" ACTING = "acting" OBSERVING = "observing" FINISHED = "finished" ERROR = "error" @dataclass class Thought: """A single thought in the agent's reasoning chain""" content: str timestamp: datetime = field(default_factory=datetime.now) action: Optional[str] = None observation: Optional[str] = None @dataclass class AgentConfig: """Configuration for agent behavior""" model: str = "claude-sonnet-4-5-20251120" max_iterations: int = 100 max_tokens_per_iteration: int = 4096 temperature: float = 0.7 tools: List[Any] = field(default_factory=list) memory_system: Optional['MemorySystem'] = None planning_enabled: bool = True reflection_enabled: bool = True class BaseAgent: """Core agent implementation""" def __init__(self, config: AgentConfig, llm_client): self.config = config self.llm = llm_client self.state = AgentState.IDLE self.thought_history: List[Thought] = [] self.tools = {tool.name: tool for tool in config.tools} async def run(self, task: str) -> Dict[str, Any]: """Main agent loop""" self.task = task self.thought_history = [] for iteration in range(self.config.max_iterations): # Think phase thought = await self.think(task) self.thought_history.append(thought) # Act phase if thought.action: result = await self.act(thought.action) thought.observation = result # Check if task is complete if self.is_complete(result): return self.format_result() else: # No action needed, provide final response return {"status": "completed", "response": thought.content} return {"status": "max_iterations", "thoughts": self.thought_history} async def think(self, task: str) -> Thought: """Reason about the current state and determine next action""" self.state = AgentState.REASONING # Build context from history and memory context = self.build_context() # Get LLM response with action response = await self.llm.chat([ {"role": "system", "content": self.get_system_prompt()}, {"role": "user", "content": f"Task: {task}\n\n{context}"} ]) return Thought(content=response.content, action=response.tool_use) async def act(self, action: str) -> str: """Execute the determined action""" self.state = AgentState.ACTING if action in self.tools: return await self.tools[action].execute() else: return action # Plain text response def build_context(self) -> str: """Build context from thought history and memory""" history = "\n".join([ f"- {t.content}" + (f" -> {t.observation}" if t.observation else "") for t in self.thought_history[-5:] ]) memory = "" if self.config.memory_system: memory = f"\nRelevant memory:\n{self.config.memory_system.retrieve(self.task)}" return f"History:\n{history}{memory}" def is_complete(self, result: str) -> bool: """Determine if task is complete""" completion_indicators = [ "task complete", "finished", "successfully", "delivered" ] return any(indicator in result.lower() for indicator in completion_indicators) def format_result(self) -> Dict[str, Any]: """Format the final result""" return { "status": "completed", "thoughts": [t.content for t in self.thought_history], "actions": [t.action for t in self.thought_history if t.action], "observations": [t.observation for t in self.thought_history if t.observation] } ``` ### ReAct Implementation ```python class ReActAgent(BaseAgent): """ReAct (Reason + Act) agent implementation""" def get_system_prompt(self) -> str: return """You are a ReAct agent. For each step: 1. Think about what to do 2. Act by calling a tool or responding 3. Observe the result Format your response as: Thought: [your reasoning] Action: [tool_name] [arguments] OR respond [your response] Observation: [result of action]""" async def think(self, task: str) -> Thought: """ReAct-style reasoning""" context = self.build_context() # Prompt for ReAct format prompt = f"""Task: {task} {context} Follow this format: Thought: [your reasoning] Action: [tool_to_use] [arguments] Observation: [result]""" response = await self.llm.chat([ {"role": "system", "content": self.get_system_prompt()}, {"role": "user", "content": prompt} ]) # Parse response return self.parse_response(response.content) def parse_response(self, response: str) -> Thought: """Parse ReAct format response""" lines = response.strip().split("\n") thought = Thought(content="") for line in lines: if line.startswith("Thought:"): thought.content = line[8:].strip() elif line.startswith("Action:"): action = line[7:].strip() if action.startswith("respond"): thought.action = action[8:].strip() else: # Parse tool call parts = action.split(" ", 1) thought.action = parts[0] if len(parts) > 1 else action elif line.startswith("Observation:"): thought.observation = line[12:].strip() return thought ``` ----- ## Tool Systems ### Tool Definition and Execution ```python from abc import ABC, abstractmethod from typing import Any, Dict import json class Tool(ABC): """Base class for agent tools""" @property @abstractmethod def name(self) -> str: """Tool name""" pass @property @abstractmethod def description(self) -> str: """Tool description for LLM""" pass @property @abstractmethod def input_schema(self) -> Dict: """JSON schema for tool input""" pass @abstractmethod async def execute(self, **kwargs) -> str: """Execute the tool""" pass class SearchTool(Tool): """Web search tool""" @property def name(self) -> str: return "search" @property def description(self) -> str: return "Search the web for information. Use this when you need current information or facts not in your training data." @property def input_schema(self) -> Dict: return { "type": "object", "properties": { "query": {"type": "string", "description": "Search query"}, "max_results": {"type": "integer", "default": 5} }, "required": ["query"] } async def execute(self, query: str, max_results: int = 5) -> str: # Implementation would call search API results = await self.search_api.search(query, max_results) return json.dumps(results) class CodeExecutionTool(Tool): """Code execution tool""" @property def name(self) -> str: return "execute_code" @property def description(self) -> str: return "Execute Python code in a sandboxed environment. Use this for calculations, data processing, or testing code snippets." @property def input_schema(self) -> Dict: return { "type": "object", "properties": { "code": {"type": "string", "description": "Python code to execute"}, "timeout": {"type": "integer", "default": 30} }, "required": ["code"] } async def execute(self, code: str, timeout: int = 30) -> str: # Would execute in sandbox result = await self.sandbox.execute(code, timeout) return str(result) class FileSystemTool(Tool): """File system operations""" @property def name(self) -> str: return "file_operations" @property def description(self) -> str: return "Read, write, or list files in the working directory." @property def input_schema(self) -> Dict: return { "type": "object", "properties": { "operation": {"type": "string", "enum": ["read", "write", "list"]}, "path": {"type": "string"}, "content": {"type": "string"} }, "required": ["operation", "path"] } async def execute(self, operation: str, path: str, content: str = None) -> str: if operation == "read": return self.read_file(path) elif operation == "write": return self.write_file(path, content) elif operation == "list": return self.list_directory(path) ``` ### Tool Manager ```python class ToolManager: """Manages available tools for the agent""" def __init__(self): self.tools: Dict[str, Tool] = {} self.tool_descriptions: List[Dict] = [] def register(self, tool: Tool): """Register a new tool""" self.tools[tool.name] = tool self.tool_descriptions.append({ "name": tool.name, "description": tool.description, "input_schema": tool.input_schema }) def get_tool(self, name: str) -> Optional[Tool]: """Get tool by name""" return self.tools.get(name) def get_descriptions(self) -> str: """Get formatted tool descriptions for LLM""" desc = "Available tools:\n" for t in self.tool_descriptions: desc += f"- {t['name']}: {t['description']}\n" desc += f" Input: {json.dumps(t['input_schema'])}\n" return desc ``` ----- ## Memory Systems ### Working Memory ```python class WorkingMemory: """Short-term memory for current task""" def __init__(self, max_items: int = 10): self.max_items = max_items self.items: List[Dict] = [] def add(self, item: Dict): """Add item to working memory""" self.items.append({ **item, "timestamp": datetime.now() }) # Keep only recent items if len(self.items) > self.max_items: self.items = self.items[-self.max_items:] def get_recent(self, n: int = 5) -> List[Dict]: """Get n most recent items""" return self.items[-n:] def clear(self): """Clear working memory""" self.items = [] def summarize(self) -> str: