LLM Orchestration Frameworks: LangChain, LlamaIndex, and DSPy

LLM Orchestration Frameworks: LangChain, LlamaIndex, and DSPy

Orchestration frameworks provide abstractions for building complex LLM applications. Each framework serves different use cases: LangChain for chains and agents, LlamaIndex for RAG, and DSPy for programmatic prompting.

Orchestration Pipeline

Framework Implementations

1. LangChain Chain Builder

from typing import Any, Callable, Dict, List, Optional
from dataclasses import dataclass
import json

@dataclass
class ChainStep:
    name: str
    function: Callable
    input_key: str
    output_key: str

class LangChainStyleChain:
    def __init__(self, name: str):
        self.name = name
        self.steps: List[ChainStep] = []
        self.memory: Dict[str, Any] = {}

    def add_step(self, name: str, function: Callable,
                 input_key: str, output_key: str):
        step = ChainStep(name=name, function=function,
                        input_key=input_key, output_key=output_key)
        self.steps.append(step)
        return self

    def run(self, initial_input: Dict) -> Dict:
        state = {**initial_input, **self.memory}
        for step in self.steps:
            try:
                input_val = state.get(step.input_key)
                output = step.function(input_val)
                state[step.output_key] = output
            except Exception as e:
                state["error"] = str(e)
                state["failed_step"] = step.name
                break
        return state

    def invoke(self, inputs: Dict) -> Dict:
        return self.run(inputs)

    def __repr__(self):
        step_names = " -> ".join([s.name for s in self.steps])
        return f"Chain({self.name}): {step_names}"

class PromptTemplate:
    def __init__(self, template: str, input_variables: List[str]):
        self.template = template
        self.input_variables = input_variables

    def format(self, **kwargs) -> str:
        result = self.template
        for key, value in kwargs.items():
            result = result.replace(f"{{{key}}}", str(value))
        return result

class OutputParser:
    @staticmethod
    def parse_json(text: str) -> Dict:
        try:
            return json.loads(text)
        except json.JSONDecodeError:
            return {"raw": text, "parse_error": True}

    @staticmethod
    def parse_list(text: str) -> List[str]:
        lines = text.strip().split("\n")
        return [line.lstrip("0123456789.- ").strip() for line in lines if line.strip()]

2. LlamaIndex RAG Pipeline

from dataclasses import dataclass
from typing import List, Dict, Optional
import numpy as np

@dataclass
class Document:
    content: str
    metadata: Dict[str, str]
    embedding: Optional[List[float]] = None

class LlamaIndexStyleRAG:
    def __init__(self, embedding_dim: int = 768):
        self.documents: List[Document] = []
        self.embedding_dim = embedding_dim
        self.index: Optional[np.ndarray] = None

    def add_documents(self, docs: List[Document]):
        for doc in docs:
            doc.embedding = self._embed(doc.content)
            self.documents.append(doc)
        self._build_index()

    def _embed(self, text: str) -> List[float]:
        return [hash(text + str(i)) % 100 / 100.0 for i in range(self.embedding_dim)]

    def _build_index(self):
        if self.documents:
            embeddings = np.array([d.embedding for d in self.documents])
            self.index = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)

    def query(self, query_text: str, top_k: int = 3) -> List[Dict]:
        query_embedding = np.array(self._embed(query_text))
        query_norm = query_embedding / np.linalg.norm(query_embedding)
        if self.index is None:
            return []
        similarities = self.index @ query_norm
        top_indices = np.argsort(similarities)[::-1][:top_k]
        results = []
        for idx in top_indices:
            results.append({
                "content": self.documents[idx].content,
                "score": float(similarities[idx]),
                "metadata": self.documents[idx].metadata
            })
        return results

    def create_index_stats(self) -> Dict:
        return {
            "num_documents": len(self.documents),
            "embedding_dim": self.embedding_dim,
            "index_size_mb": self.index.nbytes / (1024*1024) if self.index is not None else 0
        }

class QueryEngine:
    def __init__(self, rag: LlamaIndexStyleRAG, llm_caller=None):
        self.rag = rag
        self.llm = llm_caller

    def query(self, question: str, top_k: int = 3) -> str:
        context_docs = self.rag.query(question, top_k)
        context = "\n".join([d["content"] for d in context_docs])
        return f"Based on {len(context_docs)} documents, answer: {question}"

3. DSPy Programmatic Prompting

from typing import Any, Callable, Dict, List
from dataclasses import dataclass

@dataclass
class DSPySignature:
    input_fields: List[str]
    output_fields: List[str]
    instruction: str

class DSPyModule:
    def __init__(self, signature: DSPySignature):
        self.signature = signature
        self.examples: List[Dict] = []

    def forward(self, **kwargs) -> Dict:
        prompt = self._build_prompt(kwargs)
        return {"prediction": f"Output for: {list(kwargs.keys())}"}

    def _build_prompt(self, inputs: Dict) -> str:
        parts = [self.signature.instruction, ""]
        for key, val in inputs.items():
            parts.append(f"{key}: {val}")
        return "\n".join(parts)

    def fit(self, trainset: List[Dict]):
        self.examples = trainset

    def metric(self, example: Dict, prediction: Dict) -> float:
        return 1.0

class DSPyProgram:
    def __init__(self, modules: List[DSPyModule]):
        self.modules = modules

    def forward(self, **kwargs) -> Dict:
        state = kwargs
        for module in self.modules:
            result = module.forward(**state)
            state.update(result)
        return state

    def compile(self, metric: Callable = None):
        for module in self.modules:
            if self.examples:
                module.fit(self.examples)

Framework Comparison

Best Practices

1. Use LangChain for complex agent workflows with tool calling
2. Use LlamaIndex for document retrieval and RAG applications
3. Use DSPy when prompts need optimization through compilation
4. Avoid framework lock-in by keeping core logic independent
5. Benchmark overhead since frameworks add latency to simple operations