The Complete Guide to AI Agents: Architecture, Frameworks & Implementation

1. Receive task/goal
2. Plan approach (break down into steps)
3. Select appropriate tool
4. Execute tool with parameters
5. Observe result
6. Decide: goal achieved or continue loop?
7. Return final result or iterate

Thought: I need to find the current weather in Tokyo
Action: search("Tokyo weather today")
Observation: Tokyo weather is 15°C, partly cloudy
Thought: Now I have the information the user requested
Action: respond("The current weather in Tokyo is 15°C and partly cloudy")

# Pseudocode
plan = planner.create_plan(task)
# plan = ["Step 1: Search for...", "Step 2: Analyze...", "Step 3: Write..."]

for step in plan:
    result = executor.execute(step)
    if needs_replanning(result):
        plan = planner.revise_plan(task, completed_steps, result)

Manager Agent
├── Research Agent (web search, document analysis)
├── Code Agent (writing and debugging code)
├── Data Agent (database queries, data processing)
└── Writing Agent (content creation, summarization)

from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchResults
from langchain_community.utilities import WikipediaAPIWrapper
from langchain_community.tools import WikipediaQueryRun

# Define tools
search = DuckDuckGoSearchResults()
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
tools = [search, wikipedia]

# Create agent
llm = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(llm, tools, prompt_template)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Run agent
result = agent_executor.invoke({"input": "What is quantum computing?"})

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

class AgentState(TypedDict):
    messages: list
    next_action: str

# Define nodes (agent steps)
def reasoning_node(state: AgentState):
    # LLM reasoning logic
    return {"next_action": "use_tool"}

def tool_node(state: AgentState):
    # Tool execution logic
    return {"messages": state["messages"] + [tool_result]}

# Build graph
workflow = StateGraph(AgentState)
workflow.add_node("reason", reasoning_node)
workflow.add_node("tool", tool_node)
workflow.add_edge("reason", "tool")
workflow.add_conditional_edges("tool", should_continue)

agent = workflow.compile()

from crewai import Agent, Task, Crew

# Define agents with roles
researcher = Agent(
    role="Senior Research Analyst",
    goal="Research and analyze market trends",
    tools=[search_tool, scrape_tool]
)

writer = Agent(
    role="Content Writer",
    goal="Create engaging content from research",
    tools=[writing_tool]
)

# Define tasks
research_task = Task(
    description="Research AI agent frameworks",
    agent=researcher
)

writing_task = Task(
    description="Write a summary of findings",
    agent=writer
)

# Create and run crew
crew = Crew(agents=[researcher, writer], tasks=[research_task, writing_task])
result = crew.kickoff()

# pip install -U "autogen-agentchat" "autogen-ext[openai]"
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent

assistant = AssistantAgent(
    name="assistant",
    llm_config={"model": "gpt-4o"}
)

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "coding"}
)

# Start conversation
user_proxy.initiate_chat(
    assistant,
    message="Write a Python function to calculate fibonacci numbers"
)

from langchain.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> str:
    """
    Search the product database for items matching the query.

    Args:
        query: Search terms to find products
        limit: Maximum number of results to return

    Returns:
        JSON string of matching products
    """
    results = db.search(query, limit=limit)
    return json.dumps(results)

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City name"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["location"]
        }
    }
}]

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

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

# Create message history
history = InMemoryChatMessageHistory()
history.add_message(HumanMessage(content="What is Python?"))
history.add_message(AIMessage(content="Python is a programming language..."))

# Access messages
messages = history.messages

from langchain_pinecone import PineconeVectorStore
from langchain_openai import OpenAIEmbeddings
from pinecone import Pinecone

# Initialize Pinecone
pc = Pinecone(api_key="your-api-key")
index = pc.Index("agent-memory")

# Create vector store
embeddings = OpenAIEmbeddings()
vectorstore = PineconeVectorStore(index=index, embedding=embeddings)

# Add memories
vectorstore.add_texts(texts=past_interactions)

# Retrieve relevant memories
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
relevant_memories = retriever.invoke("user query")

class Episode:
    task: str
    actions: list[Action]
    result: str
    success: bool
    lessons_learned: str

from crewai import Agent, Crew, Task, Process

# Specialized agents
analyst = Agent(
    role="Data Analyst",
    goal="Analyze data and identify patterns",
    backstory="Expert in statistical analysis"
)

researcher = Agent(
    role="Research Specialist",
    goal="Gather comprehensive information",
    backstory="Skilled at finding reliable sources"
)

synthesizer = Agent(
    role="Report Writer",
    goal="Synthesize findings into clear reports",
    backstory="Expert at technical writing"
)

# Create crew with hierarchical process
crew = Crew(
    agents=[analyst, researcher, synthesizer],
    tasks=[analyze_task, research_task, write_task],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4o")
)

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def call_external_api(params):
    return api.call(params)