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
![20]()
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
![21]()
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:
Both frameworks offer powerful capabilities that can significantly improve your AI application's performance and maintainability.