Agentic AI

The ReAct Loop `` Thought → Action → Observation → Thought → Action → ... ``

Basic Agent Implementation ```python from openai import OpenAI import json

client = OpenAI()

TOOLS = [ { "type": "function", "function": { "name": "web_search", "description": "Search the web for current information", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Search query"} }, "required": ["query"] } } }, { "type": "function", "function": { "name": "run_code", "description": "Execute Python code and return the result", "parameters": { "type": "object", "properties": { "code": {"type": "string", "description": "Python code to run"} }, "required": ["code"] } } } ]

def run_agent(task: str) -> str: messages = [{"role": "user", "content": task}]

while True: response = client.chat.completions.create( model="gpt-4o", messages=messages, tools=TOOLS ) msg = response.choices[0].message messages.append(msg)

if msg.tool_calls is None: return msg.content # Agent is done

# Execute tool calls for call in msg.tool_calls: result = execute_tool(call.function.name, json.loads(call.function.arguments)) messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)}) ```

Agent Roles in a System ```python # LangGraph multi-agent workflow from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode

# Define specialized agents def researcher(state): """Agent 1: Gathers information via web search""" return search_web(state["query"])

def analyst(state): """Agent 2: Analyzes gathered data""" return analyze(state["research_results"])

def writer(state): """Agent 3: Writes the final report""" return write_report(state["analysis"])

# Build workflow workflow = StateGraph(AgentState) workflow.add_node("research", researcher) workflow.add_node("analyze", analyst) workflow.add_node("write", writer)

workflow.set_entry_point("research") workflow.add_edge("research", "analyze") workflow.add_edge("analyze", "write") workflow.add_edge("write", END)

app = workflow.compile() ```

Supervisor + Worker Pattern (CrewAI) ```python from crewai import Agent, Task, Crew

planner = Agent( role="Project Planner", goal="Break down complex tasks into subtasks for workers", backstory="Senior PM with 10 years experience" )

coder = Agent( role="Software Developer", goal="Implement features according to specifications", tools=[code_execution_tool, file_write_tool] )

reviewer = Agent( role="Code Reviewer", goal="Review code for bugs and best practices", tools=[code_execution_tool] )

crew = Crew(agents=[planner, coder, reviewer], process="hierarchical") result = crew.kickoff(inputs={"project": "Build a REST API for user auth"}) ```

Agent Memory Types ```python class AgentMemory: """ 1. Working Memory → Current conversation context (LLM context window) 2. Episodic Memory → Past task experiences (vector store of summaries) 3. Semantic Memory → Knowledge base (RAG over documents) 4. Procedural Memory → How to use tools (system prompt / examples) """

def __init__(self): self.working = [] # messages in current session self.episodic = VectorDB() # past episodes self.semantic = VectorDB() # knowledge base

def add_episode(self, task: str, outcome: str): summary = f"Task: {task}\nOutcome: {outcome}" self.episodic.add(embed(summary), {"summary": summary})

def recall_relevant(self, query: str) -> list[str]: return self.episodic.search(embed(query), top_k=3) ```

Planning Strategies ```python # Chain-of-Thought Planning plan_prompt = """ Task: {task}

Think step by step: 1. What information do I need? 2. What tools are available? 3. What is the execution order? 4. What could go wrong?

Create a numbered plan: """

# Tree-of-Thought (explore multiple paths) # 1. Generate N candidate plans # 2. Evaluate each plan's feasibility # 3. Execute the most promising plan # 4. Backtrack if plan fails ```

Safety & Guardrails ```python class SafeAgent: MAX_STEPS = 20 DANGEROUS_TOOLS = {"delete_file", "drop_database", "send_email"}

def run(self, task: str) -> str: steps = 0

while steps < self.MAX_STEPS: action = self.plan_next_action()

# Safety checks if action.tool in self.DANGEROUS_TOOLS: confirmed = self.request_human_confirmation(action) if not confirmed: return "Action cancelled by safety check"

result = self.execute(action) steps += 1

if self.is_task_complete(result): return result

return "Task exceeded maximum steps human review needed" ```

Evaluation Metrics for Agents | Metric | What it Measures | |---|---| | Task success rate | Did the agent complete the goal? | | Steps to completion | Efficiency of planning | | Tool call accuracy | Did it use the right tools? | | Hallucination rate | Did it make up tool results? | | Cost per task | Token usage × price |

Failure Modes
• Infinite loops: agent keeps replanning without making progress → add step limit
• Tool misuse: calls wrong tool or with wrong args → add tool descriptions + examples
• Context overflow: long tasks exceed context window → summarize periodically
• Cascading errors: early mistake compounds → add checkpoints and verification steps