Migrating from LangGraph to LlamaIndex/MAF

As AI applications evolve, teams often need to migrate between frameworks to leverage better capabilities, performance, or ecosystem support. This article demonstrates how to migrate a production LangGraph implementation to LlamaIndex or a Multi-Agent Framework (MAF), using a real-world customer support automation system as our use case.

Real-World Use Case: Intelligent Customer Support System

Scenario: A SaaS company needs an AI-powered customer support system that can:

Answer customer queries using product documentation
Escalate complex issues to human agents
Track conversation state and context
Integrate with multiple data sources (docs, tickets, user data)

Current LangGraph Implementation

from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, Annotated
import operator

class SupportState(TypedDict):
    messages: Annotated[list, operator.add]
    context: dict
    escalated: bool
    ticket_id: str

# Define nodes
def retrieve_context(state: SupportState):
    """Retrieve relevant documentation"""
    query = state['messages'][-1].content
    # Vector search implementation
    docs = vector_store.similarity_search(query, k=3)
    return {"context": {"docs": docs}}

def generate_response(state: SupportState):
    """Generate AI response"""
    context = state['context']['docs']
    query = state['messages'][-1].content
    
    response = llm.invoke(
        f"Context: {context}\n\nQuery: {query}\n\nProvide helpful response:"
    )
    
    return {"messages": [AIMessage(content=response)]}

def check_escalation(state: SupportState):
    """Determine if escalation is needed"""
    last_response = state['messages'][-1].content
    if "unable to help" in last_response.lower():
        return {"escalated": True}
    return {"escalated": False}

# Build graph
workflow = StateGraph(SupportState)
workflow.add_node("retrieve", retrieve_context)
workflow.add_node("generate", generate_response)
workflow.add_node("escalate", check_escalation)

workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "escalate")
workflow.add_conditional_edges(
    "escalate",
    lambda x: "end" if x['escalated'] else END
)

app = workflow.compile()

Migration Strategy

Why Migrate?

Better Data Integration: LlamaIndex excels at document indexing and retrieval
Simplified Architecture: MAF provides cleaner multi-agent patterns
Performance: Native optimizations for specific use cases
Ecosystem: Better integration with specific tools and services

Migration Approach

Phase 1: Analyze current state management and workflows
Phase 2: Map LangGraph components to LlamaIndex/MAF equivalents
Phase 3: Implement new architecture with gradual rollout
Phase 4: Testing, validation, and optimization

New Implementation with LlamaIndex

Architecture Overview

LlamaIndex Implementation

from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core.workflow import (
    StartEvent, StopEvent, Workflow, step, Event
)
from llama_index.core.agent import ReActAgent
from llama_index.llms.openai import OpenAI
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from typing import Dict, Any
import json

# Define custom events
class RetrieveEvent(Event):
    query: str
    conversation_id: str

class GenerateEvent(Event):
    query: str
    context: list
    conversation_id: str

class EscalateEvent(Event):
    query: str
    response: str
    conversation_id: str

class SupportWorkflow(Workflow):
    def __init__(self, vector_store, ticket_system, **kwargs):
        super().__init__(**kwargs)
        self.vector_store = vector_store
        self.ticket_system = ticket_system
        self.llm = OpenAI(model="gpt-4")
        self.index = VectorStoreIndex.from_vector_store(vector_store)
        
    @step
    async def start(self, ctx, ev: StartEvent) -> RetrieveEvent:
        """Entry point - receive user query"""
        query = ev.query
        conversation_id = ev.conversation_id
        
        # Store conversation state
        await ctx.set(f"conv_{conversation_id}", {
            "query": query,
            "messages": []
        })
        
        return RetrieveEvent(query=query, conversation_id=conversation_id)
    
    @step
    async def retrieve_context(self, ctx, ev: RetrieveEvent) -> GenerateEvent:
        """Retrieve relevant documentation"""
        query_engine = self.index.as_query_engine(similarity_top_k=5)
        response = query_engine.query(ev.query)
        
        # Extract source nodes as context
        context = [
            {
                "text": node.text,
                "score": node.score,
                "metadata": node.metadata
            }
            for node in response.source_nodes
        ]
        
        return GenerateEvent(
            query=ev.query,
            context=context,
            conversation_id=ev.conversation_id
        )
    
    @step
    async def generate_response(self, ctx, ev: GenerateEvent) -> EscalateEvent:
        """Generate AI response with context"""
        context_str = "\n\n".join([
            f"Doc {i+1}: {doc['text']}" 
            for i, doc in enumerate(ev.context)
        ])
        
        prompt = f"""Based on the following documentation, answer the customer query.
If you cannot find relevant information, state that clearly.

Documentation:
{context_str}

Customer Query: {ev.query}

Response:"""
        
        response = await self.llm.acomplete(prompt)
        
        return EscalateEvent(
            query=ev.query,
            response=str(response),
            conversation_id=ev.conversation_id
        )
    
    @step
    async def check_escalation(self, ctx, ev: EscalateEvent) -> StopEvent:
        """Check if escalation is needed"""
        escalation_prompt = f"""Analyze this response and determine if it needs human escalation.
Respond with JSON: {{"escalate": true/false, "reason": "explanation"}}

Response: {ev.response}"""
        
        decision = await self.llm.acomplete(escalation_prompt)
        decision_json = json.loads(str(decision))
        
        # Update conversation state
        conv_state = await ctx.get(f"conv_{ev.conversation_id}")
        conv_state["messages"].append({"role": "assistant", "content": ev.response})
        conv_state["escalated"] = decision_json.get("escalate", False)
        
        if decision_json.get("escalate"):
            # Create support ticket
            ticket_id = await self.ticket_system.create_ticket(
                query=ev.query,
                response=ev.response,
                conversation_id=ev.conversation_id
            )
            return StopEvent(result={
                "response": f"Your query has been escalated. Ticket ID: {ticket_id}",
                "escalated": True,
                "ticket_id": ticket_id
            })
        
        return StopEvent(result={
            "response": ev.response,
            "escalated": False,
            "context_used": len(ev.context)
        })

# Usage
async def main():
    # Initialize components
    vector_store = ... # Your vector store setup
    ticket_system = ... # Your ticket system setup
    
    workflow = SupportWorkflow(
        vector_store=vector_store,
        ticket_system=ticket_system,
        timeout=60,
        verbose=True
    )
    
    # Run workflow
    result = await workflow.run(
        query="How do I reset my password?",
        conversation_id="conv_123"
    )
    
    print(result)

Alternative: Multi-Agent Framework (MAF) Implementation

Architecture with Multiple Specialized Agents

MAF Implementation with CrewAI

from crewai import Agent, Task, Crew, Process
from langchain.tools import Tool
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
import json

# Define tools
def search_documentation(query: str) -> str:
    """Search product documentation"""
    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma.load_embedding_function(embeddings)
    docs = vectorstore.similarity_search(query, k=5)
    return "\n\n".join([doc.page_content for doc in docs])

def create_support_ticket(query: str, context: str) -> str:
    """Create a support ticket"""
    ticket_id = f"TKT-{hash(query) % 10000:04d}"
    # Save to database
    return json.dumps({"ticket_id": ticket_id, "status": "created"})

def analyze_complexity(query: str, response: str) -> str:
    """Analyze if query needs escalation"""
    # Simple heuristic - in production, use LLM
    complex_keywords = ["error", "broken", "urgent", "critical"]
    if any(word in query.lower() for word in complex_keywords):
        return "escalate"
    return "resolve"

# Create tools
search_tool = Tool(
    name="SearchDocumentation",
    func=search_documentation,
    description="Search product documentation for relevant information"
)

ticket_tool = Tool(
    name="CreateTicket",
    func=create_support_ticket,
    description="Create a support ticket for complex issues"
)

complexity_tool = Tool(
    name="AnalyzeComplexity",
    func=analyze_complexity,
    description="Analyze if a query needs human escalation"
)

# Define agents
orchestrator = Agent(
    role="Support Orchestrator",
    goal="Coordinate customer support workflow efficiently",
    backstory="""You are an expert at routing customer queries to the right
    agents and managing the overall support process.""",
    verbose=True,
    allow_delegation=True
)

knowledge_agent = Agent(
    role="Knowledge Specialist",
    goal="Find accurate information from documentation",
    backstory="""You excel at searching and retrieving relevant information
    from product documentation to answer customer questions.""",
    tools=[search_tool],
    verbose=True
)

response_agent = Agent(
    role="Response Generator",
    goal="Generate clear and helpful customer responses",
    backstory="""You are skilled at crafting professional, empathetic,
    and accurate responses to customer queries.""",
    verbose=True
)

escalation_agent = Agent(
    role="Escalation Specialist",
    goal="Identify and handle complex issues requiring human intervention",
    backstory="""You can detect when a query is too complex for AI handling
    and create appropriate support tickets.""",
    tools=[ticket_tool, complexity_tool],
    verbose=True
)

# Define tasks
def create_support_tasks(query: str):
    retrieval_task = Task(
        description=f"""Search documentation for information about: {query}
        Provide relevant excerpts that can help answer this query.""",
        expected_output="Relevant documentation excerpts",
        agent=knowledge_agent
    )
    
    response_task = Task(
        description=f"""Based on the retrieved information, generate a helpful
        response to the customer query: {query}
        
        Make it professional, clear, and actionable.""",
        expected_output="Customer response",
        agent=response_agent,
        context=[retrieval_task]
    )
    
    escalation_task = Task(
        description=f"""Analyze the query and response to determine if escalation
        is needed. If yes, create a support ticket.
        
        Query: {query}
        Response: {{response_task.output}}""",
        expected_output="Escalation decision and ticket ID if needed",
        agent=escalation_agent,
        context=[response_task]
    )
    
    return [retrieval_task, response_task, escalation_task]

# Create crew
support_crew = Crew(
    agents=[orchestrator, knowledge_agent, response_agent, escalation_agent],
    tasks=create_support_tasks("How do I reset my password?"),
    process=Process.sequential,
    verbose=True
)

# Execute
result = support_crew.kickoff()
print(result)

Migration Comparison

Feature Mapping

LangGraph Component	LlamaIndex Equivalent	MAF Equivalent
StateGraph	Workflow	Crew
State (TypedDict)	Context (ctx)	Shared Memory
Nodes	Steps (@step)	Agents
Edges	Event Flow	Task Dependencies
Conditional Edges	Event Routing	Agent Delegation
Checkpointing	Workflow Persistence	Memory Management

Performance Considerations

# Benchmarking script
import time
import asyncio
from typing import List

async def benchmark_workflow(workflow, queries: List[str], iterations: int = 10):
    """Benchmark workflow performance"""
    times = []
    
    for _ in range(iterations):
        for query in queries:
            start = time.time()
            result = await workflow.run(query=query, conversation_id="test")
            end = time.time()
            times.append(end - start)
    
    return {
        "avg_time": sum(times) / len(times),
        "min_time": min(times),
        "max_time": max(times),
        "total_queries": len(times)
    }

# Usage
queries = [
    "How do I reset my password?",
    "What are the pricing plans?",
    "How to integrate with API?",
    "Why is my account locked?",
    "How to export data?"
]

# Benchmark LangGraph
langgraph_time = await benchmark_workflow(langgraph_app, queries)

# Benchmark LlamaIndex
llamaindex_time = await benchmark_workflow(llamaindex_workflow, queries)

print(f"LangGraph: {langgraph_time['avg_time']:.2f}s avg")
print(f"LlamaIndex: {llamaindex_time['avg_time']:.2f}s avg")

Best Practices for Migration

1. Gradual Migration Strategy

class HybridWorkflow:
    """Gradually migrate from LangGraph to LlamaIndex"""
    
    def __init__(self, use_new: bool = False):
        self.use_new = use_new
        self.old_workflow = LangGraphWorkflow()
        self.new_workflow = LlamaIndexWorkflow()
    
    async def process(self, query: str):
        if self.use_new:
            return await self.new_workflow.run(query=query)
        else:
            return await self.old_workflow.run(query=query)

# Feature flag for gradual rollout
USE_NEW_WORKFLOW = os.getenv("USE_NEW_WORKFLOW", "false").lower() == "true"
hybrid = HybridWorkflow(use_new=USE_NEW_WORKFLOW)

2. State Migration

def migrate_state(langgraph_state: dict) -> dict:
    """Convert LangGraph state to LlamaIndex format"""
    return {
        "messages": langgraph_state.get("messages", []),
        "context": langgraph_state.get("context", {}),
        "metadata": {
            "escalated": langgraph_state.get("escalated", False),
            "ticket_id": langgraph_state.get("ticket_id", ""),
            "migrated_from": "langgraph",
            "migration_timestamp": datetime.now().isoformat()
        }
    }

Deployment Configuration

Docker Compose for Production

version: '3.8'

services:
  # API Service
  api:
    build: ./api
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - VECTOR_DB_URL=http://vector-db:6333
      - POSTGRES_URL=postgresql://user:pass@postgres:5432/support
    depends_on:
      - vector-db
      - postgres
    deploy:
      replicas: 3

  # Vector Database (Qdrant)
  vector-db:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
    volumes:
      - vector_data:/qdrant/storage

  # PostgreSQL
  postgres:
    image: postgres:15
    environment:
      POSTGRES_DB: support
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
    volumes:
      - postgres_data:/var/lib/postgresql/data

  # Redis for caching
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  # Worker for async tasks
  worker:
    build: ./worker
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
    deploy:
      replicas: 2

volumes:
  vector_data:
  postgres_data:

Migrating from LangGraph to LlamaIndex or MAF provides several advantages:

Better Data Handling: LlamaIndex's native document processing capabilities
Cleaner Multi-Agent Patterns: MAF frameworks provide more intuitive agent coordination
Improved Performance: Optimized for specific use cases
Richer Ecosystem: Better integration with modern AI tools

Key Takeaways

Start with a clear migration strategy and phased approach
Maintain backward compatibility during transition
Implement comprehensive testing at each stage
Monitor performance and quality metrics
Document the new architecture thoroughly

The choice between LlamaIndex and MAF depends on your specific needs:

Choose LlamaIndex if your focus is on document-heavy applications with complex retrieval needs
Choose MAF (CrewAI) if you need sophisticated multi-agent coordination with clear role separation

Both frameworks offer powerful capabilities that can significantly improve your AI application's performance and maintainability.