LangChainチュートリアル: 基礎から高度なRAGまで

# Uncomment to install core LangChain and provider-specific packages
# !pip install langchain langchain-core langchain-community langchain-openai langchain-chroma faiss-cpu pypdf sentence-transformers tiktoken -q

# Uncomemnt to install environment management and web request libraries
# !pip install python-dotenv requests beautifulsoup4 -q

import os
import getpass
from dotenv import load_dotenv

# Load environment variables from a .env file if it exists
load_dotenv()

# Set up your OpenAI API key
if "OPENAI_API_KEY" not in os.environ:
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

print("✅ API keys configured!")

from langchain.chat_models import init_chat_model

# Initialize a chat model from OpenAI
model = init_chat_model("gpt-4o-mini", model_provider="openai", temperature=0.7)

print(f"✅ Initialized model: {model.__class__.__name__}")

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Create a prompt template
prompt_template = ChatPromptTemplate.from_messages([
    ("system", "You are an expert in {expertise}."),
    ("human", "Explain {topic} in a simple and concise way.")
])

# Create a simple chain using LCEL
# The chain will:
# 1. Take user input for 'expertise' and 'topic'.
# 2. Format the prompt using the prompt_template.
# 3. Pass the formatted prompt to the model.
# 4. Parse the model's output to a string.
simple_chain = prompt_template | model | StrOutputParser()

# Invoke the chain
response = simple_chain.invoke({
    "expertise": "physics",
    "topic": "quantum entanglement"
})

print(response)

print("🌊 Streaming response:")
for chunk in simple_chain.stream({
    "expertise": "culinary arts",
    "topic": "the Maillard reaction"
}):
    print(chunk, end="", flush=True)

from typing import List
from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser

# Define a Pydantic model for structured output
class Recipe(BaseModel):
    name: str = Field(description="The name of the recipe")
    ingredients: List[str] = Field(description="A list of ingredients")
    steps: List[str] = Field(description="The steps to prepare the recipe")

# Create a Pydantic output parser
pydantic_parser = PydanticOutputParser(pydantic_object=Recipe)

# Create a prompt that includes format instructions
structured_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a world-class chef. Generate a recipe based on the user's request and format it as requested."),
    ("human", "I want a simple recipe for {dish}.\n\n{format_instructions}")
])

# Create the structured output chain
structured_chain = structured_prompt | model | pydantic_parser

# Invoke the chain
recipe_request = {
    "dish": "scrambled eggs",
    "format_instructions": pydantic_parser.get_format_instructions()
}
recipe_output = structured_chain.invoke(recipe_request)

print(f"Recipe for: {recipe_output.name}")
print("\nIngredients:")
for ingredient in recipe_output.ingredients:
    print(f"- {ingredient}")

from langchain_core.runnables import RunnableBranch

# Define different prompts for different levels
beginner_prompt = ChatPromptTemplate.from_template("Explain {topic} to a complete beginner within 60 words.")
expert_prompt = ChatPromptTemplate.from_template("Provide a detailed, technical explanation of {topic} within 60 words.")

# Create a conditional chain using RunnableBranch
# This chain checks the 'level' input and routes to the appropriate prompt
conditional_chain = (
    RunnableBranch(
        (lambda x: x.get("level") == "expert", expert_prompt),
        beginner_prompt  # Default prompt
    )
    | model.bind(max_tokens=100)
    | StrOutputParser()
 )

# Test the beginner path
beginner_response = conditional_chain.invoke({"topic": "black holes", "level": "beginner"})
print("--- Beginner Explanation ---")
print(beginner_response)

# Test the expert path
expert_response = conditional_chain.invoke({"topic": "black holes", "level": "expert"})
print("\n--- Expert Explanation ---")
print(expert_response)

from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Load documents from a web page
loader = WebBaseLoader("https://python.langchain.com/docs/modules/model_io/prompts/")
docs = loader.load()

# Initialize a text splitter
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)

# Split the documents into chunks
splits = text_splitter.split_documents(docs)

print(f"Loaded {len(docs)} documents and split them into {len(splits)} chunks.")

# Preview the first few chunks
print("\n📖 Preview of chunked text:")
for i, chunk in enumerate(splits[:3]):  # Show first 3 chunks
    print(f"\n--- Chunk {i+1} ---")
    print(f"Content: {chunk.page_content[:200]}...")  # Show first 200 characters
    print(f"Metadata: {chunk.metadata}")
    print(f"Full length: {len(chunk.page_content)} characters")

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
import numpy as np

# Initialize OpenAI embeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Preview embeddings - let's see what embeddings look like
sample_text = "LangChain is a framework for developing applications powered by language models."
sample_embedding = embeddings.embed_query(sample_text)

print("📊 Embedding Preview:")
print(f"Sample text: '{sample_text}'")
print(f"Embedding dimension: {len(sample_embedding)}")
print(f"Embedding type: {type(sample_embedding)}")
print(f"First 10 values: {sample_embedding[:10]}")
print(f"Embedding range: [{min(sample_embedding):.4f}, {max(sample_embedding):.4f}]")
print(f"Embedding norm: {np.linalg.norm(sample_embedding):.4f}")

# Create a Chroma vector store from the document splits
vectorstore = Chroma.from_documents(documents=splits, embedding=embeddings)

print(f"\n✅ Vector store created with {len(splits)} document chunks.")
print("Each chunk has been converted to a {}-dimensional embedding vector.".format(len(sample_embedding)))

# Create a retriever from the vector store
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# Test the retriever
query = "What are prompt templates?"
retrieved_docs = retriever.invoke(query)

print(f"Retrieved {len(retrieved_docs)} documents for the query: '{query}'")

from langchain_core.runnables import RunnablePassthrough

# Define a RAG prompt template
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Use the following context to answer the user's question.\n\nContext:\n{context}"),
    ("human", "{question}")
])

# Helper function to format the retrieved documents
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# Create the RAG chain
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | model
    | StrOutputParser()
)

# Test the RAG chain
rag_question = "How can I use few-shot examples in my prompts?"
rag_answer = rag_chain.invoke(rag_question)

print(f"\n❓ Question: {rag_question}")
print(f"🎯 Answer: {rag_answer}")