2026年版 データサイエンティストのためのLangChain: LLM、エージェント、面接対策

# langchain_setup.py
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage

# Initialize with specific model and temperature
openai_model = ChatOpenAI(
    model="gpt-4o",
    temperature=0.1,  # Lower temperature for deterministic outputs
    max_tokens=2048
)

# Anthropic model with same interface
anthropic_model = ChatAnthropic(
    model="claude-sonnet-4-20250514",
    temperature=0.1
)

# Both models accept identical message format
messages = [
    SystemMessage(content="You are a data science expert."),
    HumanMessage(content="Explain gradient boosting in one sentence.")
]

# Swap models without changing message structure
response = openai_model.invoke(messages)

# lcel_chain.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from pydantic import BaseModel, Field

# Define structured output schema
class DataInsight(BaseModel):
    metric: str = Field(description="The metric being analyzed")
    trend: str = Field(description="Upward, downward, or stable")
    confidence: float = Field(description="Confidence score 0-1")

# Create prompt template with variables
prompt = ChatPromptTemplate.from_messages([
    ("system", "Analyze the following data and extract insights."),
    ("human", "Dataset: {dataset_description}\nFocus on: {metric}")
])

# LCEL chain with pipe operator
model = ChatOpenAI(model="gpt-4o", temperature=0)
parser = JsonOutputParser(pydantic_object=DataInsight)

chain = prompt | model | parser  # Pipe syntax composes runnables

# Invoke with input dictionary
result = chain.invoke({
    "dataset_description": "Monthly sales data showing 15% increase",
    "metric": "revenue growth"
})
# Returns: DataInsight(metric='revenue growth', trend='upward', confidence=0.85)

# rag_pipeline.py
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

# Initialize embeddings and vector store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
    collection_name="data_docs",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

# Create retriever with relevance threshold
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.7, "k": 4}
)

# RAG prompt with context injection
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """Answer based on the provided context only.
Context: {context}
If the context doesn't contain relevant information, say so."""),
    ("human", "{question}")
])

# Format documents into context string
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# RAG chain with parallel retrieval
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | ChatOpenAI(model="gpt-4o", temperature=0)
)

# Query with automatic retrieval
answer = rag_chain.invoke("What normalization method works best for skewed distributions?")

# react_agent.py
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
import pandas as pd
import numpy as np

# Define data analysis tools
@tool
def calculate_statistics(data: str, column: str) -> str:
    """Calculate descriptive statistics for a column in CSV data."""
    df = pd.read_csv(pd.io.common.StringIO(data))
    stats = df[column].describe()
    return stats.to_string()

@tool
def detect_outliers(data: str, column: str, method: str = "iqr") -> str:
    """Detect outliers using IQR or Z-score method."""
    df = pd.read_csv(pd.io.common.StringIO(data))
    values = df[column]
    
    if method == "iqr":
        q1, q3 = values.quantile([0.25, 0.75])
        iqr = q3 - q1
        outliers = values[(values < q1 - 1.5 * iqr) | (values > q3 + 1.5 * iqr)]
    else:  # z-score
        z_scores = np.abs((values - values.mean()) / values.std())
        outliers = values[z_scores > 3]
    
    return f"Found {len(outliers)} outliers: {outliers.tolist()[:10]}"

@tool
def correlation_analysis(data: str, col1: str, col2: str) -> str:
    """Calculate Pearson correlation between two columns."""
    df = pd.read_csv(pd.io.common.StringIO(data))
    corr = df[col1].corr(df[col2])
    return f"Correlation between {col1} and {col2}: {corr:.4f}"

# Create ReAct agent with tools
model = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [calculate_statistics, detect_outliers, correlation_analysis]

agent = create_react_agent(
    model=model,
    tools=tools,
    state_modifier="You are a data analyst. Use tools to analyze data step by step."
)

# Agent reasons about which tools to use
result = agent.invoke({
    "messages": [("human", "Analyze the sales column for outliers and basic stats")]
})

# memory_patterns.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# Initialize model and prompt with history placeholder
model = ChatOpenAI(model="gpt-4o", temperature=0.1)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a data science assistant helping with analysis."),
    MessagesPlaceholder(variable_name="history"),  # Inject conversation history
    ("human", "{input}")
])

chain = prompt | model

# Session-based message store
message_stores = {}

def get_session_history(session_id: str):
    if session_id not in message_stores:
        message_stores[session_id] = InMemoryChatMessageHistory()
    return message_stores[session_id]

# Wrap chain with message history
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# Conversation maintains context across invocations
config = {"configurable": {"session_id": "analysis_session_1"}}

response1 = chain_with_history.invoke(
    {"input": "I'm analyzing customer churn data with 50k rows"},
    config=config
)
# Later in the same session
response2 = chain_with_history.invoke(
    {"input": "What sampling strategy should I use for this dataset size?"},
    config=config
)  # Agent remembers the 50k rows context

# streaming_example.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("Explain {concept} in detail")
model = ChatOpenAI(model="gpt-4o", streaming=True)

chain = prompt | model

# Stream yields chunks as they arrive
for chunk in chain.stream({"concept": "backpropagation"}):
    print(chunk.content, end="", flush=True)

# caching_setup.py
from langchain_core.globals import set_llm_cache
from langchain_community.cache import RedisCache
import redis

# Redis cache for distributed deployments
redis_client = redis.Redis(host="localhost", port=6379)
set_llm_cache(RedisCache(redis_client))

# Subsequent identical queries hit cache
model = ChatOpenAI(model="gpt-4o")
response1 = model.invoke("What is gradient descent?")  # API call
response2 = model.invoke("What is gradient descent?")  # Cache hit