AI Agent Development Complete Guide [2026] - From Concept to Practice

# ReAct pattern example
def react_loop(agent, goal):
    while not goal_achieved:
        # Think: analyze current state
        thought = agent.think(current_state)
        # Act: select and execute tool
        action = agent.select_action(thought)
        # Observe: get execution result
        observation = agent.execute(action)
        # Update state
        current_state = update_state(observation)

# Vector database implementation for long-term memory
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

class AgentMemory:
    def __init__(self):
        self.short_term = []  # Recent N conversation turns
        self.long_term = Chroma(embedding_function=OpenAIEmbeddings())
    
    def store(self, content, memory_type="short"):
        if memory_type == "short":
            self.short_term.append(content)
            if len(self.short_term) > 10:
                self.short_term.pop(0)
        else:
            self.long_term.add_texts([content])
    
    def retrieve(self, query, k=5):
        return self.long_term.similarity_search(query, k=k)

from langchain.tools import Tool, tool

@tool
def search_web(query: str) -> str:
    """Search the web for latest information"""
    # Implement search logic
    return search_results

@tool  
def execute_python(code: str) -> str:
    """Execute Python code and return result"""
    # Safely execute code
    return exec_result

tools = [search_web, execute_python]

def reflect(agent, task, result):
    reflection_prompt = f"""
    Task: {task}
    Execution Result: {result}
    
    Please analyze:
    1. Does the result fully meet the task requirements?
    2. What aspects of the execution process can be optimized?
    3. If re-executing, how should the strategy be adjusted?
    """
    return agent.llm.invoke(reflection_prompt)

from langchain.agents import AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain import hub

llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)

tools = [
    Tool(
        name="Search",
        func=lambda q: search_api(q),
        description="Search the web for information"
    ),
    Tool(
        name="Calculator",
        func=lambda expr: eval(expr),
        description="Perform mathematical calculations"
    )
]

prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({
    "input": "Query today's weather in Shanghai and calculate the Fahrenheit temperature"
})

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role='Researcher',
    goal='Collect and analyze market data',
    backstory='You are an experienced market research expert',
    tools=[search_tool, scrape_tool],
    llm=llm
)

analyst = Agent(
    role='Analyst',
    goal='Generate insight reports based on research data',
    backstory='You are a data analysis expert skilled at identifying trends',
    tools=[analysis_tool],
    llm=llm
)

writer = Agent(
    role='Writer',
    goal='Transform analysis results into readable reports',
    backstory='You are a professional business writing expert',
    llm=llm
)

research_task = Task(
    description='Research 2024 AI Agent market trends',
    agent=researcher,
    expected_output='Market data and key findings'
)

analysis_task = Task(
    description='Analyze market data and identify major trends',
    agent=analyst,
    expected_output='Trend analysis report'
)

report_task = Task(
    description='Write the final market analysis report',
    agent=writer,
    expected_output='Complete market analysis report'
)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.sequential
)

result = crew.kickoff()

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

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_step: str

def plan_step(state: AgentState):
    # Plan next action
    return {"next_step": "execute", "messages": ["Planning complete"]}

def execute_step(state: AgentState):
    # Execute action
    return {"next_step": "reflect", "messages": ["Execution complete"]}

def reflect_step(state: AgentState):
    # Reflect on results
    if task_complete:
        return {"next_step": "end", "messages": ["Task complete"]}
    return {"next_step": "plan", "messages": ["Need to re-plan"]}

workflow = StateGraph(AgentState)

workflow.add_node("plan", plan_step)
workflow.add_node("execute", execute_step)
workflow.add_node("reflect", reflect_step)

workflow.set_entry_point("plan")
workflow.add_edge("plan", "execute")
workflow.add_edge("execute", "reflect")
workflow.add_conditional_edges(
    "reflect",
    lambda x: x["next_step"],
    {"plan": "plan", "end": END}
)

app = workflow.compile()

system_prompt = """
You are a professional task execution Agent.

## Working Principles
1. Analyze the task and create a clear plan before taking action
2. Execute only one tool call at a time
3. Carefully observe tool return results
4. If results don't meet expectations, reflect on reasons and adjust strategy
5. After completing the task, summarize the execution process

## Available Tools
{tools_description}

## Output Format
Thought: [Analyze current state and next step plan]
Action: [Selected tool and parameters]
"""

class RobustAgent:
    def __init__(self, max_retries=3):
        self.max_retries = max_retries
    
    def execute_with_retry(self, task):
        for attempt in range(self.max_retries):
            try:
                result = self.execute(task)
                if self.validate_result(result):
                    return result
                # Result doesn't meet requirements, reflect and retry
                self.reflect_and_adjust(task, result)
            except Exception as e:
                self.handle_error(e, attempt)
        return self.fallback_response(task)