この記事は、2024-12-13 の TORICO の技術勉強会の内容を元に作成しました。

前提説明

LangChain とは

テキスト生成AIを使うための便利なツールキット(フレームワーク)

🦜🔗 の絵文字で表される。

バージョン 0.3になって、以前の使用方法の多くが Deprecated になった。(1.0で廃止) 書籍を買うのはおすすめしない。ブログ記事も2024年年初のものは古い。 公式サイトを読むのが一番混乱が少ない。

この記事は 2024年12月に書いたが、2025年下旬には古くて使えなくなっている可能性がある。

LangSmith

https://smith.langchain.com/

LangChain の GoogleAnalytics 的なもの。 LLM の呼び出し内容、応答時間、RAG で参考にしたドキュメントなどが確認できます。

今回のチュートリアルで使うので、Github でアカウントを作ってください。

手順

プロジェクトフォルダの作成

mkdir langchain-tutorial

環境構築

今回のプロジェクトの範囲外のライブラリもインストールしています。PDF ローダーやチャットUI，エージェントライブラリなど。

今回使わなかったライブラリは次回使います。

a. Pipenv を使う場合

Pipfile

[[source]]
name = "pypi"
url = "https://pypi.org/simple"
verify_ssl = true

[dev-packages]

[packages]
# 基本セット
langchain = "*"
openai = "*"
python-dotenv = "*"
langchain-community = "*"
langchain-openai = "*"
langgraph = "*"

# PDFの読み込み
pymupdf = "*"
spacy = "*"

# ChromaDB RAG
tiktoken = "*"
chromadb = "*"
langchain-chroma = "*"

# チャットUI
chainlit = "*"

# Agent
wikipedia = "*"
playwright = "*"

# LLMMathChain(算術計算)
numexpr = "*"

# API Server
fastapi = "*"
uvicorn = "*"
langserve = "*"

[requires]
python_version = "3.12"

pipenv install

b. uv を使う場合

pyproject.toml

[project]
name = "langchain-tutorial"
version = "0.1.0"
requires-python = ">=3.12,<3.13"
dependencies = [
    "langchain",
    "openai",
    "python-dotenv",
    "langchain-community",
    "langchain-openai",
    "langgraph",

    # PDFの読み込み
    "pymupdf",
    "spacy",

    # ChromaDB RAG
    "tiktoken",
    "chromadb",
    "langchain-chroma",

    # チャットUI
    "chainlit",

    # Agent
    "wikipedia",
    "playwright",

    # LLMMathChain(算術計算)
    "numexpr",

    # API Server
    "fastapi",
    "uvicorn",
    "langserve",
]

uv sync

仮想環境に入る

. .venv/bin/activate

.env ファイルを作る

.env

OPENAI_API_KEY=sk-...

# LANGSMITH
LANGCHAIN_TRACING_V2=true
LANGCHAIN_ENDPOINT="https://api.smith.langchain.com"
LANGCHAIN_API_KEY="lsv2_pt_..."
# LANGCHAIN_PROJECT="pr-..."

OPENAI_API_KEY には所有している OpenAI の APIキーを入れる。

このチュートリアルの最初に作った、 LangSmith のページの左下の歯車マークから、 API キーを作成し、 LANGCHAIN_API_KEY に設定する。

LANGCHAIN_PROJECT は、今回はコメントアウトでOK

scripts フォルダを作る

mkdir scripts

今回のチュートリアルで使うスクリプトは、この scripts フォルダの中に書いていく。

現在のフォルダ構成

Pipenv で環境構築した場合

📁 langchain-tutorial
    + Pipfile
    + Pipfile.lock
    + .env
    + 📁 scripts

uv で環境構築した場合

📁 langchain-tutorial
    + pyproject.toml
    + uv.lock
    + .env
    + 📁 scripts

シンプルな LLM

scripts/t01_model_invoke.py

# https://python.langchain.com/docs/tutorials/llm_chain/

import dotenv
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

llm = ChatOpenAI(model='gpt-4o-mini')

answer = llm.invoke(
    [
        SystemMessage(content='入力された日本の会社の特徴を答えなさい'),
        HumanMessage(content='株式会社TORICO'),
    ]
)

print(answer)

scripts/t02_output_parser.py

OutputParser を使って、回答のオブジェクトから本文のみを取得する。

# https://python.langchain.com/docs/tutorials/llm_chain/

import dotenv
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

parser = StrOutputParser()
dotenv.load_dotenv()

llm = ChatOpenAI(model='gpt-4o-mini')

answer = llm.invoke(
    [
        SystemMessage(content='入力されたEコマースサービスの特徴を答えなさい'),
        HumanMessage(content='まんが王'),
    ]
)

print(parser.invoke(answer))

LangSmith の確認

https://smith.langchain.com/

先ほどの LLM コールが記録されている。

Runnable チェーン を使う

scripts/t03_chain.py

import dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()


prompt_template = ChatPromptTemplate.from_messages(
    [
        ('system', '入力された{category}の特徴を答えなさい'),
        ('user', '{text}'),
    ]
)
# print(prompt_template.invoke({'category': '日本の会社', 'text': '株式会社TORICO'}))
chat_model = ChatOpenAI(model='gpt-4o-mini')
output_parser = StrOutputParser()

# RunnableSerializable のインスタンスをチェーンして RunnableSequence を作る
chain = prompt_template | chat_model | output_parser

# ChatPromptValue
answer = chain.invoke({'category': '日本のECサイト', 'text': 'マンガ展'})

print(answer)

answer = chain.invoke({'category': '日本のECサイト', 'text': 'トレオタ'})

print(answer)

解説

chain = prompt_template | chat_model | output_parser

が注目箇所。

ChatPromptTemplate, ChatOpenAI, StrOutputParser の Type Hierarchy を見てみると、 RunnableSerializable を継承しているのがわかる。

つまり prompt_template, chat_model, output_parser は、それぞれ RunnableSerializable のサブクラスのインスタンス。

LangChain Ver3 以降は、 RunnableSerializable のインスタンスを | でチェーンして、 .invoke() するのが基本的な使い方となる。

APIサーバー

scripts/t04_langserve.py

FastAPI を使って、HTTP APIサーバーを簡単に構築できる。

#!/usr/bin/env python3
import dotenv
from fastapi import FastAPI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langserve import add_routes

dotenv.load_dotenv()


prompt_template = ChatPromptTemplate.from_messages(
    [
        (
            'system',
            '次の{category}の特徴を答えなさい。'
            '不明な場合は、過剰な推測はせず、素直にわからないと答えてください。',
        ),
        ('user', '{text}'),
    ]
)

chat_model = ChatOpenAI(model='gpt-4o-mini')

output_parser = StrOutputParser()

chain = prompt_template | chat_model | output_parser

app = FastAPI(
    title='LangChain Serve',
    description='LangServe is a language model serving API.',
    version='0.1.0',
)

# Langserve でつなげる
add_routes(
    app,
    chain,
    path='/chain',
)

if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host='localhost', port=8000)
    # open http://localhost:8000/chain/playground/

内部で Pydantic でデータのバリデーションをしている。

このスクリプトの中でも、先ほどと同じく

chain = prompt_template | chat_model | output_parser

の箇所で RunnableSerializable をチェーンして RunnableSequence を作っている。

それを、 langserve.add_routes を使って FastAPI の API エンドポイントに繋げている。

もし

ImportError: cannot import name 'VerifyTypes' from 'httpx._types'. Did you mean: 'ProxyTypes'?

が出た場合、httpx をダウングレードする。

python3 pip install httpx==0.27.2
python3 pip install sse-starlette

uv add httpx==0.27.2
uv add sse-starlette

APIクライアント

scripts/t05_remote_runnable.py

from langserve import RemoteRunnable

remote_chain = RemoteRunnable("http://localhost:8000/chain/")
response = remote_chain.invoke({"category": "日本の地名", "text": "九段下"})

print(response)

先ほどの APIサーバーを起動したまま、今回のスクリプトを実行する。

短いコードで API コールが出来る。

text を自分の生誕地に変えてみて実行する。

RAG + メモリーセーバー

今回のチュートリアルのまとめとして、以下の処理を行うスクリプトを書く。

1. Wikipedia の「ジョジョの奇妙な冒険」に関する記事を何ページか取得する
2. それをいくつかのドキュメントに分割する
3. 分割したドキュメントを OpenAI の API を使ってベクトル解析し、LangChain にビルトインされているベクターストアDBに格納する。
4. 利用者の質問を入力し、ベクターストアを検索し、結果をRAG で返す。会話履歴は LangChain にビルトインされているメモリーセーバーに記録する。
5. 利用者からの質問をさらに入力する。前回の会話履歴を元に、適した回答を作る。

scripts/t23_rag_memory_saver_ja.py

from typing import Sequence

import bs4
import dotenv
from langchain.chains import (
    create_history_aware_retriever,
    create_retrieval_chain,
)
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
from typing_extensions import Annotated, TypedDict

dotenv.load_dotenv()


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


### Construct retriever ###


url_paths = [
    "https://ja.wikipedia.org/wiki/%E3%82%B8%E3%83%A7%E3%82%B8%E3%83%A7%E3%81%AE%E5%A5%87%E5%A6%99%E3%81%AA%E5%86%92%E9%99%BA",
    # # 3部
    "https://ja.wikipedia.org/wiki/%E3%82%B9%E3%82%BF%E3%83%BC%E3%83%80%E3%82%B9%E3%83%88%E3%82%AF%E3%83%AB%E3%82%BB%E3%82%A4%E3%83%80%E3%83%BC%E3%82%B9",
    # # 4部
    # "https://ja.wikipedia.org/wiki/%E3%83%80%E3%82%A4%E3%83%A4%E3%83%A2%E3%83%B3%E3%83%89%E3%81%AF%E7%A0%95%E3%81%91%E3%81%AA%E3%81%84",
    # # 5部
    # "https://ja.wikipedia.org/wiki/%E9%BB%84%E9%87%91%E3%81%AE%E9%A2%A8",
]

loader = WebBaseLoader(
    web_paths=url_paths,
    # BeautifulSoup を使って本文だけを抽出したい時のコード
    # 結果が不安定だったため今回は使わなかった。
    # bs_kwargs=dict(
    #     parse_only=bs4.SoupStrainer(
    #         class_=("post-content", "post-title", "post-header")
    #     )
    # ),
)
docs = loader.load()

# テキストスプリッター。今回は雑にサイズ 1000 で分割。
# 本来は HTML 構造を解析するのが良い
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000, chunk_overlap=200
)
splits = text_splitter.split_documents(docs)

# print(splits)

vectorstore = InMemoryVectorStore.from_documents(
    documents=splits, embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()


### Contextualize question ###
contextualize_q_system_prompt = (
    "チャット履歴と、チャット履歴のコンテキストを参照する可能性のある最新のユーザー質問が与えられた場合、"
    "チャット履歴がなくても理解できる独立した質問を作成します。"
    "質問には回答せず、必要に応じて質問を作り直し、それ以外の場合はそのまま返します。"
)
contextualize_q_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", contextualize_q_system_prompt),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}"),
    ]
)
history_aware_retriever = create_history_aware_retriever(
    llm, retriever, contextualize_q_prompt
)


### Answer question ###
system_prompt = (
    "ユーザーからの質問に対して、「検索結果コンテキスト」の内容を元に回答してください。"
    "答えがわからない場合は、わからないと答えてください。"
    "\n\n"
    "# 検索結果コンテキスト\n{context}"
)

qa_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system_prompt),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}"),
    ]
)
question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)

rag_chain = create_retrieval_chain(
    history_aware_retriever, question_answer_chain
)


### Statefully manage chat history ###
class State(TypedDict):
    input: str
    chat_history: Annotated[Sequence[BaseMessage], add_messages]
    context: str
    answer: str


def call_model(state: State):
    response = rag_chain.invoke(state)
    return {
        "chat_history": [
            HumanMessage(state["input"]),
            AIMessage(response["answer"]),
        ],
        "context": response["context"],
        "answer": response["answer"],
    }


workflow = StateGraph(state_schema=State)
workflow.add_edge(START, "model")
workflow.add_node("model", call_model)

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)


# ------
# thread_id は適当な文字列を指定する
config = {"configurable": {"thread_id": "abc123"}}

result = app.invoke(
    {"input": "時間を止めるスタンドは何?"},
    config=config,
)
print(result["answer"])

result = app.invoke(
    {"input": "その本体の名前は?"},
    config=config,
)
print(result["answer"])

# いろいろ書いてみよう

実行し、結果を確認したら、LangSmith を見る。

https://smith.langchain.com/

RAGの際、どのようなドキュメントが選択されたかなど見れる。