# Helixdevelopment Helixagent Langchain Migration Deep Dive

> LangChain Migration Deep Dive

- Skill: `tomevault-io/helixdevelopment-helixagent-langchain-migration-deep-dive` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add tomevault-io/helixdevelopment-helixagent-langchain-migration-deep-dive`
- Raw SKILL.md: https://api.skillmd.com/api/skills/tomevault-io/helixdevelopment-helixagent-langchain-migration-deep-dive/raw
- Safety review: pending (external: skill-scanner WARNING, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: tomevault-io (https://skillmd.com/u/tomevault-io)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/tomevault-io/helixdevelopment-helixagent-langchain-migration-deep-dive

---


# LangChain Migration Deep Dive

## Overview
Comprehensive strategies for migrating to LangChain from legacy LLM implementations or other frameworks.

## Prerequisites
- Existing LLM application to migrate
- Understanding of current architecture
- Test coverage for validation
- Staging environment for testing

## Migration Scenarios

### Scenario 1: Raw OpenAI SDK to LangChain

#### Before (Raw SDK)
```python
# legacy_openai.py
import openai

client = openai.OpenAI()

def chat(message: str, history: list = None) -> str:
    messages = [{"role": "system", "content": "You are helpful."}]

    if history:
        messages.extend(history)

    messages.append({"role": "user", "content": message})

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        temperature=0.7
    )

    return response.choices[0].message.content
```

#### After (LangChain)
```python
# langchain_chat.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import HumanMessage, AIMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are helpful."),
    MessagesPlaceholder(variable_name="history"),
    ("user", "{message}")
])

chain = prompt | llm | StrOutputParser()

def chat(message: str, history: list = None) -> str:
    # Convert legacy format to LangChain messages
    lc_history = []
    if history:
        for msg in history:
            if msg["role"] == "user":
                lc_history.append(HumanMessage(content=msg["content"]))
            elif msg["role"] == "assistant":
                lc_history.append(AIMessage(content=msg["content"]))

    return chain.invoke({"message": message, "history": lc_history})
```

### Scenario 2: LlamaIndex to LangChain

#### Before (LlamaIndex)
```python
# legacy_llamaindex.py
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.llms.openai import OpenAI

documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(llm=OpenAI(model="gpt-4o-mini"))

def query(question: str) -> str:
    response = query_engine.query(question)
    return str(response)
```

#### After (LangChain)
```python
# langchain_rag.py
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import DirectoryLoader
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# Load documents
loader = DirectoryLoader("data")
documents = loader.load()

# Split documents
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits = splitter.split_documents(documents)

# Create vector store
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(splits, embeddings)
retriever = vectorstore.as_retriever()

# Create RAG chain
llm = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_template("""
Answer based on the context:

Context: {context}

Question: {question}
""")

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

def query(question: str) -> str:
    return chain.invoke(question)
```

### Scenario 3: Custom Agent to LangChain Agent

#### Before (Custom)
```python
# legacy_agent.py
import json

def run_agent(query: str, tools: dict) -> str:
    messages = [{"role": "user", "content": query}]

    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            functions=[{"name": k, **v["schema"]} for k, v in tools.items()]
        )

        msg = response.choices[0].message

        if msg.function_call:
            # Execute tool
            tool_name = msg.function_call.name
            tool_args = json.loads(msg.function_call.arguments)
            result = tools[tool_name]["func"](**tool_args)

            messages.append({"role": "function", "name": tool_name, "content": result})
        else:
            return msg.content
```

#### After (LangChain)
```python
# langchain_agent.py
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool

# Convert tools to LangChain format
@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Results for: {query}"

@tool
def calculate(expression: str) -> str:
    """Calculate a math expression."""
    return str(eval(expression))

tools = [search, calculate]

llm = ChatOpenAI(model="gpt-4o")

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant with tools."),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

def run_agent(query: str) -> str:
    result = executor.invoke({"input": query})
    return result["output"]
```

## Migration Strategy

### Phase 1: Assessment
```python
# migration_assessment.py
import ast
import os
from pathlib import Path
from dataclasses import dataclass
from typing import List

@dataclass
class MigrationItem:
    file: str
    line: int
    pattern: str
    complexity: str  # low, medium, high

def assess_codebase(directory: str) -> List[MigrationItem]:
    """Scan codebase for migration items."""
    items = []
    patterns = {
        "openai.ChatCompletion": ("OpenAI SDK v0", "medium"),
        "openai.OpenAI": ("OpenAI SDK v1", "low"),
        "llama_index": ("LlamaIndex", "high"),
        "langchain.chains": ("LangChain legacy chains", "medium"),
        "LLMChain": ("Legacy LLMChain", "low"),
    }

    for path in Path(directory).rglob("*.py"):
        with open(path) as f:
            content = f.read()
            for i, line in enumerate(content.split("\n"), 1):
                for pattern, (name, complexity) in patterns.items():
                    if pattern in line:
                        items.append(MigrationItem(
                            file=str(path),
                            line=i,
                            pattern=name,
                            complexity=complexity
                        ))

    return items

# Generate migration report
items = assess_codebase("src/")
print(f"Found {len(items)} migration items:")
for item in items:
    print(f"  {item.file}:{item.line} - {item.pattern} ({item.complexity})")
```

### Phase 2: Parallel Implementation
```python
# Run both systems in parallel for validation
class DualRunner:
    """Run legacy and new implementations side by side."""

    def __init__(self, legacy_fn, new_fn):
        self.legacy_fn = legacy_fn
        self.new_fn = new_fn
        self.discrepancies = []

    async def run(self, *args, **kwargs):
        """Run both and compare."""
        legacy_result = await self.legacy_fn(*args, **kwargs)
        new_result = await self.new_fn(*args, **kwargs)

        if not self._compare(legacy_result, new_result):
            self.discrepancies.append({
                "args": args,
                "kwargs": kwargs,
                "legacy": legacy_result,
                "new": new_result
            })

        # Return new implementation result
        return new_result

    def _compare(self, a, b) -> bool:
        """Compare results for equivalence."""
        # Implement comparison logic
        return True  # Placeholder
```

### Phase 3: Gradual Rollout
```python
# Feature flag based rollout
import random

class FeatureFlag:
    """Control rollout percentage."""

    def __init__(self, rollout_percentage: float = 0):
        self.percentage = rollout_percentage

    def is_enabled(self, user_id: str = None) -> bool:
        """Check if feature is enabled for user."""
        if user_id:
            # Consistent per-user
            hash_val = hash(user_id) % 100
            return hash_val < self.percentage
        return random.random() * 100 < self.percentage

# Usage
langchain_flag = FeatureFlag(rollout_percentage=10)  # 10% rollout

def process_request(user_id: str, message: str):
    if langchain_flag.is_enabled(user_id):
        return langchain_chat(message)
    else:
        return legacy_chat(message)
```

### Phase 4: Validation and Cleanup
```python
# Validation script
import pytest

class MigrationValidator:
    """Validate migration is complete and correct."""

    def __init__(self, test_cases: list):
        self.test_cases = test_cases

    def run_validation(self, new_fn) -> dict:
        """Run all test cases and report."""
        results = {"passed": 0, "failed": 0, "errors": []}

        for case in self.test_cases:
            try:
                result = new_fn(**case["input"])
                if self._validate(result, case["expected"]):
                    results["passed"] += 1
                else:
                    results["failed"] += 1
                    results["errors"].append({
                        "case": case,
                        "actual": result
                    })
            except Exception as e:
                results["failed"] += 1
                results["errors"].append({
                    "case": case,
                    "error": str(e)
                })

        return results

    def _validate(self, actual, expected) -> bool:
        """Validate result meets expectations."""
        # Implement validation logic
        return True

# Run validation
validator = MigrationValidator([
    {"input": {"message": "Hello"}, "expected": {"type": "greeting"}},
    # ... more test cases
])

results = validator.run_validation(langchain_chat)
print(f"Passed: {results['passed']}, Failed: {results['failed']}")
```

## Migration Checklist
- [ ] Codebase assessed for migration items
- [ ] Test coverage added for current behavior
- [ ] LangChain equivalents implemented
- [ ] Parallel running validation passed
- [ ] Gradual rollout completed
- [ ] Legacy code removed
- [ ] Documentation updated

## Common Issues

| Issue | Solution |
|-------|----------|
| Different response format | Add output parser adapter |
| Missing streaming support | Implement streaming callbacks |
| Memory format mismatch | Convert message history format |
| Tool schema differences | Update tool definitions |

## Resources
- [LangChain Migration Guide](https://python.langchain.com/docs/versions/migrating_chains/)
- [OpenAI SDK Migration](https://github.com/openai/openai-python/discussions/742)
- [Feature Flags Best Practices](https://launchdarkly.com/blog/best-practices-feature-flags/)

## Next Steps
Use `langchain-upgrade-migration` for LangChain version upgrades.

---
> Source: [HelixDevelopment/HelixAgent](https://github.com/HelixDevelopment/HelixAgent) — distributed by [TomeVault](https://tomevault.io).
<!-- tomevault:4.0:skill_md:2026-06-16 -->