Agentic AI for Enterprise: Building Autonomous AI Systems in 2026

# ReAct Loop
while not task_complete:
    thought = llm.reason(observation, goal, history)
    action = llm.select_action(thought, available_tools)
    observation = execute_tool(action)
    history.append((thought, action, observation))
    
    if should_escalate(observation):
        await human_approval()

# Plan-and-Execute
plan = planner_agent.create_plan(goal, context)

for step in plan.steps:
    result = executor_agent.execute(step)
    
    if result.requires_replanning:
        plan = planner_agent.revise_plan(plan, result)
    
    plan.mark_complete(step, result)

# Multi-Agent Orchestration
agents = {
    "researcher": ResearchAgent(),
    "analyst": AnalysisAgent(),
    "writer": WritingAgent(),
    "reviewer": QualityAgent(),
}

async def orchestrate(task):
    research = await agents["researcher"].gather_info(task)
    analysis = await agents["analyst"].analyse(research)
    draft = await agents["writer"].compose(analysis)
    final = await agents["reviewer"].review_and_refine(draft)
    return final

# Install LangGraph
pip install langgraph langchain-openai langchain-anthropic

# Or with all extras
pip install "langgraph[all]"

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.postgres import PostgresSaver
import operator

# Define state schema
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    task_status: str
    requires_approval: bool
    audit_log: list

# Define tools with enterprise guardrails
@tool
def query_database(query: str, database: str) -> str:
    """Execute read-only SQL query against approved databases."""
    approved_dbs = ["analytics", "reporting", "public_data"]
    if database not in approved_dbs:
        return f"Error: Database '{database}' not in approved list"
    # Execute query with read-only connection
    return execute_readonly_query(query, database)

@tool
def send_notification(recipient: str, message: str, channel: str) -> str:
    """Send notification via approved channels."""
    if channel not in ["email", "slack", "teams"]:
        return "Error: Unsupported notification channel"
    return send_message(recipient, message, channel)

@tool
def create_ticket(title: str, description: str, priority: str) -> str:
    """Create support ticket in ITSM system."""
    return create_jira_ticket(title, description, priority)

# Configure LLM with function calling
llm = ChatOpenAI(
    model="gpt-4-turbo-preview",
    temperature=0,
).bind_tools([query_database, send_notification, create_ticket])

# Define graph nodes
def should_continue(state: AgentState) -> str:
    """Determine next step based on agent state."""
    last_message = state["messages"][-1]
    
    if state["requires_approval"]:
        return "await_approval"
    if last_message.tool_calls:
        return "tools"
    return "end"

def call_model(state: AgentState) -> AgentState:
    """Invoke LLM with current state."""
    messages = state["messages"]
    response = llm.invoke(messages)
    
    # Check if action requires human approval
    requires_approval = check_approval_required(response)
    
    return {
        "messages": [response],
        "requires_approval": requires_approval,
        "audit_log": [{
            "action": "llm_response",
            "content": response.content,
            "timestamp": datetime.utcnow().isoformat()
        }]
    }

def await_human_approval(state: AgentState) -> AgentState:
    """Pause for human approval of high-impact actions."""
    # In production, this would trigger a notification
    # and wait for approval via webhook or polling
    approval = request_human_approval(state)
    
    if approval.approved:
        return {
            "requires_approval": False,
            "audit_log": [{
                "action": "human_approval",
                "approved_by": approval.approver,
                "timestamp": datetime.utcnow().isoformat()
            }]
        }
    else:
        return {
            "messages": [AIMessage(content="Action cancelled by human reviewer")],
            "task_status": "cancelled"
        }

# Build the graph
workflow = StateGraph(AgentState)

workflow.add_node("agent", call_model)
workflow.add_node("tools", ToolNode([query_database, send_notification, create_ticket]))
workflow.add_node("await_approval", await_human_approval)

workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "await_approval": "await_approval",
        "end": END
    }
)

workflow.add_edge("tools", "agent")
workflow.add_edge("await_approval", "tools")

# Compile with persistence
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/agents"
)
app = workflow.compile(checkpointer=checkpointer)

# Execute agent with thread persistence
config = {"configurable": {"thread_id": "enterprise-task-001"}}

result = app.invoke(
    {
        "messages": [HumanMessage(
            content="Analyse Q4 sales data and send a summary to the leadership team"
        )],
        "task_status": "pending",
        "requires_approval": False,
        "audit_log": []
    },
    config
)

# Resume from checkpoint if interrupted
state = app.get_state(config)
print(f"Current status: {state.values['task_status']}")

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
from langchain_openai import ChatOpenAI

# Configure LLM
llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0.1)

# Define specialised agents
market_researcher = Agent(
    role="Senior Market Research Analyst",
    goal="Gather comprehensive market intelligence and competitive analysis",
    backstory="""You are an expert market analyst with 15 years of experience 
    in technology sector research. You excel at identifying market trends, 
    competitive positioning, and growth opportunities.""",
    tools=[SerperDevTool(), ScrapeWebsiteTool()],
    llm=llm,
    verbose=True,
    allow_delegation=True
)

financial_analyst = Agent(
    role="Financial Analysis Expert",
    goal="Analyse financial data and provide investment recommendations",
    backstory="""You are a CFA charterholder with expertise in equity research 
    and financial modelling. You provide data-driven financial assessments.""",
    llm=llm,
    verbose=True
)

strategy_advisor = Agent(
    role="Strategic Planning Director",
    goal="Synthesise research into actionable strategic recommendations",
    backstory="""You are a former McKinsey partner who excels at translating 
    complex analysis into clear, actionable business strategies.""",
    llm=llm,
    verbose=True
)

# Define tasks
market_research_task = Task(
    description="""Research the AI infrastructure market for 2026:
    1. Identify top 5 competitors in the GPU cloud space
    2. Analyse pricing trends and market positioning
    3. Document emerging technologies and disruption risks
    4. Assess market size and growth projections""",
    expected_output="Comprehensive market research report with data tables",
    agent=market_researcher
)

financial_analysis_task = Task(
    description="""Based on the market research, analyse:
    1. Revenue projections for key players
    2. Margin analysis and cost structures
    3. Investment requirements for market entry
    4. ROI scenarios for different strategies""",
    expected_output="Financial analysis with projections and sensitivity analysis",
    agent=financial_analyst,
    context=[market_research_task]
)

strategy_task = Task(
    description="""Synthesise the research and analysis into:
    1. Executive summary (1 page)
    2. Strategic options with pros/cons
    3. Recommended approach with rationale
    4. Implementation roadmap with milestones""",
    expected_output="Strategic recommendation document ready for board presentation",
    agent=strategy_advisor,
    context=[market_research_task, financial_analysis_task]
)

# Create and run the crew
analysis_crew = Crew(
    agents=[market_researcher, financial_analyst, strategy_advisor],
    tasks=[market_research_task, financial_analysis_task, strategy_task],
    process=Process.sequential,  # Or Process.hierarchical for manager pattern
    verbose=True
)

result = analysis_crew.kickoff()
print(result)

from langchain_core.tools import tool
from pydantic import BaseModel, Field
from typing import Optional
import structlog

logger = structlog.get_logger()

class DatabaseQueryInput(BaseModel):
    """Input schema for database query tool."""
    query: str = Field(description="SQL query to execute (SELECT only)")
    database: str = Field(description="Target database name")
    timeout_seconds: int = Field(default=30, description="Query timeout")
    
class DatabaseQueryOutput(BaseModel):
    """Output schema for database query tool."""
    success: bool
    data: Optional[list] = None
    row_count: int = 0
    error: Optional[str] = None
    execution_time_ms: int = 0

@tool(args_schema=DatabaseQueryInput, return_direct=False)
def secure_database_query(
    query: str,
    database: str,
    timeout_seconds: int = 30
) -> DatabaseQueryOutput:
    """
    Execute a read-only SQL query against an approved enterprise database.
    
    This tool enforces:
    - Read-only queries (SELECT only)
    - Approved database whitelist
    - Query timeout limits
    - Row count limits for safety
    
    Returns structured output with success status and data.
    """
    # Validate database access
    approved_databases = get_approved_databases_for_agent()
    if database not in approved_databases:
        logger.warning("database_access_denied", 
                      database=database, 
                      approved=approved_databases)
        return DatabaseQueryOutput(
            success=False,
            error=f"Access denied: '{database}' not in approved list"
        )
    
    # Validate query is read-only
    if not is_read_only_query(query):
        logger.warning("write_query_blocked", query=query[:100])
        return DatabaseQueryOutput(
            success=False,
            error="Only SELECT queries are permitted"
        )
    
    # Execute with timeout and row limit
    try:
        start_time = time.time()
        result = execute_with_timeout(
            query=query,
            database=database,
            timeout=timeout_seconds,
            max_rows=10000
        )
        execution_time = int((time.time() - start_time) * 1000)
        
        logger.info("query_executed",
                   database=database,
                   rows=len(result),
                   time_ms=execution_time)
        
        return DatabaseQueryOutput(
            success=True,
            data=result,
            row_count=len(result),
            execution_time_ms=execution_time
        )
        
    except TimeoutError:
        return DatabaseQueryOutput(
            success=False,
            error=f"Query timed out after {timeout_seconds}s"
        )
    except Exception as e:
        logger.error("query_failed", error=str(e))
        return DatabaseQueryOutput(
            success=False,
            error=f"Query failed: {str(e)}"
        )

from langchain_openai import OpenAIEmbeddings
from langchain_postgres import PGVector
from datetime import datetime

class AgentMemory:
    """Enterprise agent memory system with multiple memory types."""
    
    def __init__(self, agent_id: str, user_id: str):
        self.agent_id = agent_id
        self.user_id = user_id
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        
        # Long-term memory store
        self.long_term = PGVector(
            collection_name=f"memory_{agent_id}",
            connection=DATABASE_URL,
            embeddings=self.embeddings
        )
        
        # Episodic memory for task history
        self.episodic = PGVector(
            collection_name=f"episodes_{agent_id}",
            connection=DATABASE_URL,
            embeddings=self.embeddings
        )
    
    async def remember(self, content: str, memory_type: str, metadata: dict = None):
        """Store new memory with metadata."""
        doc_metadata = {
            "user_id": self.user_id,
            "timestamp": datetime.utcnow().isoformat(),
            "memory_type": memory_type,
            **(metadata or {})
        }
        
        store = self.long_term if memory_type == "knowledge" else self.episodic
        await store.aadd_texts([content], metadatas=[doc_metadata])
    
    async def recall(self, query: str, memory_type: str = "all", k: int = 5) -> list:
        """Retrieve relevant memories."""
        filters = {"user_id": self.user_id}
        if memory_type != "all":
            filters["memory_type"] = memory_type
        
        results = await self.long_term.asimilarity_search(
            query, k=k, filter=filters
        )
        return results
    
    async def get_relevant_context(self, task: str) -> str:
        """Build context from relevant memories for a task."""
        # Recall relevant knowledge
        knowledge = await self.recall(task, memory_type="knowledge", k=3)
        
        # Recall similar past tasks
        episodes = await self.recall(task, memory_type="episode", k=2)
        
        context_parts = []
        if knowledge:
            context_parts.append("Relevant knowledge:\n" + 
                               "\n".join([d.page_content for d in knowledge]))
        if episodes:
            context_parts.append("Similar past tasks:\n" + 
                               "\n".join([d.page_content for d in episodes]))
        
        return "\n\n".join(context_parts)