What Is Agentic AI? A Practical Guide for Enterprise Engineering Teams

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import tool
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferWindowMemory
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# --- Tool Definitions ---
@tool
def query_customer_database(customer_id: str) -> str:
    """Look up a customer record by customer ID. Returns account status, tier, and recent activity."""
    # In production: replace with actual DB query
    mock_data = {
        "C-1042": {"name": "Acme Corp", "tier": "Enterprise", "status": "active", "arr": 120000, "last_activity": "2025-06-01"},
        "C-0891": {"name": "Globex Ltd", "tier": "Pro", "status": "at-risk", "arr": 48000, "last_activity": "2025-04-15"},
    }
    record = mock_data.get(customer_id)
    if not record:
        return f"No customer found with ID {customer_id}"
    return str(record)

@tool
def query_support_tickets(customer_id: str, days: int = 30) -> str:
    """Retrieve open and recently closed support tickets for a customer in the last N days."""
    # In production: replace with actual helpdesk API call
    mock_tickets = {
        "C-0891": [
            {"id": "T-4421", "status": "open", "priority": "high", "subject": "API latency spikes", "created": "2025-05-20"},
            {"id": "T-4388", "status": "closed", "priority": "medium", "subject": "Billing discrepancy", "created": "2025-04-30"}
        ]
    }
    tickets = mock_tickets.get(customer_id, [])
    if not tickets:
        return f"No support tickets found for {customer_id} in the last {days} days"
    return str(tickets)

@tool
def draft_escalation_email(customer_id: str, account_name: str, issue_summary: str) -> str:
    """Draft a customer success escalation email. Does NOT send — returns the draft for human review."""
    draft = f"""Subject: Urgent: Account Review Required — {account_name}

Dear Customer Success Manager,

This is an automated flag from the CS Agent for account {customer_id} ({account_name}).

Summary: {issue_summary}

Recommended action: Schedule an urgent call with the account owner within 48 hours.

This draft requires your review and approval before sending."""
    logger.info(f"Draft escalation created for {customer_id}")
    return draft

# --- Agent Setup ---
tools = [query_customer_database, query_support_tickets, draft_escalation_email]

llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,
    max_tokens=4096
)

react_prompt = PromptTemplate.from_template("""
You are an enterprise Customer Success Agent. Your job is to investigate customer accounts,
identify at-risk customers, and prepare escalation recommendations for the CS team.

You have access to the following tools:
{tools}

Available tool names: {tool_names}

Use this format:
Thought: what you need to do and why
Action: the tool name
Action Input: the input to the tool
Observation: the result
... (repeat Thought/Action/Action Input/Observation as needed)
Thought: I now have enough information to provide a final answer
Final Answer: your complete analysis and recommendation

Current conversation history:
{chat_history}

User request: {input}
{agent_scratchpad}
""")

memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    k=5,  # keep last 5 exchanges
    return_messages=False
)

agent = create_react_agent(llm=llm, tools=tools, prompt=react_prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True,
    max_iterations=8,
    max_execution_time=120,  # 2 minute hard limit
    handle_parsing_errors=True,
    return_intermediate_steps=True
)

def run_cs_agent(request: str) -> dict:
    """Run the customer success agent and return structured results."""
    try:
        result = agent_executor.invoke({"input": request})
        return {
            "success": True,
            "output": result["output"],
            "steps_taken": len(result.get("intermediate_steps", []))
        }
    except Exception as e:
        logger.error(f"Agent execution failed: {e}")
        return {"success": False, "error": str(e)}

if __name__ == "__main__":
    result = run_cs_agent(
        "Investigate customer C-0891 and tell me if they are at risk of churning. "
        "If they are, draft an escalation email for the CS team."
    )
    print(f"\nAgent completed in {result.get('steps_taken', 0)} steps")
    print(result["output"])