LangChain - 闘うITエンジニアの覚え書き

• TOP
• Python覚え書き
• LangChain

目次 †

• 関連
  • LangServe
  • LangSmith
  • LangGraph
  • プロンプトエンジニアリング

概要 †

規模言語モデル（LLM）を使ったアプリケーション開発を効率化するためのフレームワークで、
「プロンプトの管理」「外部データとの連携」「複雑な推論フローの構築」を容易にする事を目的とする。

基本的な構成要素

準備 †

requirements.txt

openai==1.40.6
langchain-core==0.3.58
langchain-openai==0.2.0
pydantic==2.10.6
langchain-community==0.3.0
langchain-text-splitters==0.3.8
langchain-chroma==0.1.4
GitPython==3.1.43

実行時に「Client.__init__() got an unexpected keyword argument 'proxies'」エラーになる場合は httpx のバージョンを落とす。

!pip uninstall -y httpx
!pip install httpx==0.27.2

依存ライブラリインストール

pip install -r requirements.txt

環境変数の設定

os.environ["OPENAI_API_KEY"] = "OpenAIのAPIキー"

主な構成要素 †

langchain_core.prompts.ChatPromptTemplate †

LLM（大規模言語モデル）に渡すプロンプトのテンプレートを定義・構築するクラス。
変数（例：{question}）を埋め込み、入力内容に応じて動的にプロンプト文を生成できる。

[用途]

• チャット形式の会話テンプレートを作成
• 入力変数を受け取ってプロンプト文を動的に整形
• 一貫したフォーマットでモデルに指示を送る

[使用例]

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("質問: {question}\n回答:")
formatted = prompt.format(question="LangChainとは何ですか？")
print(formatted)

langchain_openai.ChatOpenAI †

OpenAI の Chat モデル（例：GPT-4 など）を扱うためのラッパークラス。
LangChain の統一インターフェースを通じて、ChatGPTスタイルの応答を取得できる。

[用途]

• OpenAIのチャットモデルにメッセージを送信
• ストリーミング出力や温度（temperature）などのパラメータ制御
• LangChainパイプラインの一部として利用（例：prompt | model | parser）

[使用例]

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
response = llm.invoke("こんにちは、自己紹介して。")
print(response.content)

langchain_core.output_parsers.StrOutputParser †

モデルからの出力を文字列として単純に抽出するためのパーサ。
LLM出力をJSONや構造化データにせず、そのままテキストで扱いたい場合に使用。

[用途]

• チャット応答をテキストとして取り出す
• 複雑な構文解析が不要な場合に使用

[使用例]

from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()
result = parser.invoke({"text": "Hello world"})
print(result)  # "Hello world"

典型的なパイプライン †

ここではパイプ演算子を使用しない場合/使用する場合のコードをそれぞれ記載する

準備 †

プロンプト、モデル、パーサーを以下の通り準備する。

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

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "ユーザーが入力したソフトウェアについて簡潔に説明して下さい。"),
        ("human", "{keyword}"),
    ]
)
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
output_parser = StrOutputParser()

パイプ演算を使用しない場合 †

ChatPromptTemplate、ChatOpenAI、StrOutputParser はすべて LangChainの「Runnable」という抽象基底クラスを継承しており、
このクラスの invoke メソッドのアウトプットは、そのまま次の Runnable のインプットにする事で一連の処理をパイプラインとして実行する事が出来る。

prompt_value = prompt.invoke({"keyword": "docker"})
ai_message = model.invoke(prompt_value)
output = output_parser.invoke(ai_message)

print(output)

パイプ演算を使用してチェインする場合 †

Runnableを | で繋ぐ事によりパイプラインを実行する事が出来る。
※ | でつなぐと「RunnableSequence」になるが、これも Runnable の一種。

chain = prompt | model | output_parser
output = chain.invoke({"keyword": "docker"})
print(output)

streamの利用 †

stream を利用する事で、返却された結果をリアルタイムに順次取得する事が出来る。

chain = prompt | model | output_parser

for chunk in chain.stream({"keyword": "docker"}):
    print(chunk, end="", flush=True)

batchの利用 †

batch を利用して同じパイプラインを複数のパラメータが異なるで実行する事が出来る。
※非同期処理用の ainvoke・astream・abatch メソッドも存在する

chain = prompt | model | output_parser

outputs = chain.batch([{"keyword": "docker"}, {"keyword": "jupyter"}])
for i, line in enumerate(outputs):
    print(f"----- [{i}] -----")
    print(line)

任意の関数をRunnable にする †

RunnableLambda を使用すれば任意の関数をRunnableにする事が出来る。

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableLambda

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "ユーザーが入力したソフトウェアについて簡潔に説明して下さい。"),
        ("human", "{keyword}について"),
    ]
)
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
output_parser = StrOutputParser()

def upper(text: str) -> str:
    return text.upper()

chain = prompt | model | output_parser | RunnableLambda(upper)
output = chain.invoke({"keyword": "Rust"})
print(output)

RunnableParallelによる並列処理 †

複数のRunnableを並列につなげるには RunnableParallel を使用する

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

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

optimistic_prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたは楽観主義者です。ユーザーの入力に対して楽観的な意見をください。"),
    ("human", "{topic}"),
])
optimistic_chain = optimistic_prompt | model | output_parser

pessimistic_prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたは悲観主義者です。ユーザーの入力に対して悲観的な意見をください。"),
    ("human", "{topic}"),
])
pessimistic_chain = pessimistic_prompt | model | output_parser

import pprint
from langchain_core.runnables import RunnableParallel

parallel_chain = RunnableParallel({"optimistic_opinion": optimistic_chain, "pessimistic_opinion": pessimistic_chain })
output = parallel_chain.invoke({"topic": "生成AIの進化について"})

pprint.pprint(output)

RunnableParallelの出力をRunnableの入力に連結する †

synthesize_prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたは客観的AIです。2つの意見をまとめてください。"),
    ("human", "楽観的意見: {optimistic_opinion}\n悲観的意見:{pessimistic_opinion}"),
])
synthesize_chain = (
    RunnableParallel(
        {
            "optimistic_opinion": optimistic_chain,
            "pessimistic_opinion": pessimistic_chain,
        }
    )
    | synthesize_prompt
    | model
    | output_parser
)
output = synthesize_chain.invoke({"topic": "Goについて"})
print(output)

itemgetter を利用して ChatPromptTemplate への穴埋めも可能 †

from operator import itemgetter
:

synthesize_prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたは客観的AIです。{topic} 2つの意見をまとめてください。"),
    ("human", "楽観的意見: {optimistic_opinion}\n悲観的意見: {pessimistic_opinion}"),
])
synthesize_chain2 = (
    {
        "optimistic_opinion": optimistic_chain,
        "pessimistic_opinion": pessimistic_chain,
        "topic": itemgetter("topic"),
    }
    | synthesize_prompt
    | model
    | output_parser
)
output2 = synthesize_chain2.invoke({"topic": "クリーンアーキテクチャについて"})
print(output2)

その他のOutputParser †

StrOutputParser 以外の OutputParser についても触れておく。

SimpleJsonOutputParser †

from langchain.output_parsers.json import SimpleJsonOutputParser
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# LLM
model = ChatOpenAI(model="gpt-4o-mini")

# ChatPromptTemplate を使用
prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたは JSON を生成するアシスタントです。"),
    (
        "human",
        """
ユーザーの基本情報を JSON で返してください。
ユーザの自己紹介文からユーザの基本情報を抽出してJSONで返して下さい
必ず以下の属性は含める事として、紹介文から抽出できない場合は null として下さい。

必ず含める属性
username: ユーザ名
age: 年齢
sex: 性別

ユーザーの自己紹介分: {userinfo}
"""
    )
])

# 出力パーサ
parser = SimpleJsonOutputParser()

# Runnable のパイプライン化
chain = prompt | model | parser

# 実行
result = chain.invoke({"userinfo": "私の名前は山田太郎です。20歳です"})

print(result, type(result))
for k in result:
    print(f"{k}: {result[k]}")

結果

{'username': '山田太郎', 'age': 20, 'sex': None} 
username: 山田太郎
age: 20
sex: None

BooleanOutputParser †

from langchain.output_parsers import BooleanOutputParser
from langchain_openai import ChatOpenAI

parser = BooleanOutputParser()
model = ChatOpenAI(model="gpt-4o-mini")

prompt = "以下の文は肯定的ですか？ Yes か No で答えてください: 今日は良い日だ。"

result = (model | parser).invoke(prompt)
print(result, type(result))  # True or False

結果

True 

DatetimeOutputParser †

from langchain.output_parsers import DatetimeOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

parser = DatetimeOutputParser(format="%Y-%m-%d")
model = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたはユーザーの文章から正確な日付だけを抽出するアシスタントです。"),
    ("human",
    """
本日は 2025/1/1 です。
次の文から会議の日付を判定して、
必ず日付のみを YYYY-MM-DD の形式で返してください。

文: {text}
""")
])

chain = prompt | model | parser

res = chain.invoke({"text": "明後日会議があります。"})
print(res, type(res))

結果

2025-01-03 00:00:00 

EnumOutputParser †

from enum import Enum
from langchain.output_parsers import EnumOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

class Sentiment(Enum):
    POSITIVE = "positive"
    NEGATIVE = "negative"

parser = EnumOutputParser(enum=Sentiment)
model = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたは文の感情を判定し、結果を strict に返すアシスタントです。"),
    ("human",
    """
次の文の感情を判定し、
以下のいずれか **1語のみ** を返してください。

- positive
- negative

説明文は禁止。引用符も付けず、余計な文字も出力してはいけません。

文: {text}
""")
])

chain = prompt | model | parser

res = chain.invoke({"text": "今日の仕事がとても楽しかった"})
print(res, type(res))

結果

Sentiment.NEGATIVE 

PydanticOutputParser †

from pydantic import BaseModel
from langchain.output_parsers import PydanticOutputParser
from langchain_openai import ChatOpenAI

class UserInfo(BaseModel):
    name: str
    age: int
    hobbies: list[str]

parser = PydanticOutputParser(pydantic_object=UserInfo)
model = ChatOpenAI(model="gpt-4o-mini")

prompt = f"""
次の Pydantic モデルに従って JSON を出力してください:

{parser.get_format_instructions()}
"""

res = (model | parser).invoke(prompt)
print(res)
print(type(res))  # UserInfo

結果

name='Alice' age=30 hobbies=['reading', 'traveling', 'gaming']

CommaSeparatedListOutputParser †

from langchain.output_parsers import CommaSeparatedListOutputParser
from langchain_openai import ChatOpenAI

#parser = StrOutputParser()
parser = CommaSeparatedListOutputParser()
model = ChatOpenAI(model="gpt-4o-mini")

prompt = """
以下の文章から重要なキーワードを抽出して、
コンマ区切りで出力してください。

東京株式市場は6日続伸を記録しました。
日銀金融政策決定会合の結果、政策金利は予想通り変更されず（0.5%）、物価と経済成長の見通しのみが下方修正されたことが背景にあります。
日銀の利上げが後ずれするとの見方から、日経平均株価は一時36,500円台を回復しました
"""

chain = model | parser
res = chain.invoke(prompt)
print(res, type(res))

結果

['東京株式市場', '6日続伸', '日銀金融政策決定会合', '政策金利', '変更なし', '物価', '経済成長', '下方修正', '利上げ', '日経平均株価', '36', '500円台', '回復'] 

会話履歴の保持（ConversationMemory） †

会話型アプリで過去の文脈を保持するために使われる。

ツールとエージェント †

Tool 及び Agent は LLM に「行動（Action）」の選択をさせるために利用出来る。
ツールを使って目標を達成する「思考 + 行動 + 観察」を自動で行うエージェントを作成する事が出来る。

前知識 †

LLM は以下のステップを繰り返す

• Thought（推論） → 何をすべきか考える
• Action（行動） → 必要ならツールを実行（電卓、API、DB検索など）
• Observation（観察） → ツールの結果を受け取る
• 再び Thought に戻る → 得た情報を元に次の行動を決める

このループを続けて、最後に Final Answer（結論） を返す

その他（用語など） †

• ReAct
  ReAct（Reason + Act）はLLM が「推論（Reason）」と「行動（Act）」を交互に行うことで、複雑なタスクを段階的に解くための思考フレームワークの事
• Xxxxx
  TODO:

ツールとエージェントの使用例(1) †

(注) バージョン 0.1.0 以降は initialize_agent は非推奨(※)
※ create_react_agent, create_json_agent, create_structured_chat_agent を使えとの事（ https://api.python.langchain.com/en/latest/agents/langchain.agents.initialize.initialize_agent.html ）

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import initialize_agent, AgentType


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


agent = initialize_agent(
    tools=[calculator],
    llm=ChatOpenAI(model="gpt-4o-mini"),
    agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION  # 行動する前に推論ステップを実行する最も基本的なゼロショットエージェント
)

ツールとエージェントの使用例(2) †

create_react_agent で書き直してみる

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub

# --- ツール定義 ---
@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression."""
    return str(eval(expression))

# --- LLM ---
llm = ChatOpenAI(model="gpt-4o-mini")

# --- ReAct 用プロンプトを LangChain Hub から取得 ---
prompt = hub.pull("hwchase17/react")   # ← ReAct 用テンプレート

# --- ReAct Agent を生成 ---
agent = create_react_agent(
    llm=llm,
    tools=[calculator],
    prompt=prompt,
)

# --- 実行ラッパー（AgentExecutor） ---
executor = AgentExecutor(
    agent=agent,
    tools=[calculator],
    verbose=True
)

# --- 実行 ---
result = executor.invoke({"input": "What is 3 * (2 + 4)?"})
print(result["output"])

ツールとエージェントの使用例(3) †

さらに Hub に繋ぐ必要がないようにローカルプロンプトで書き直してみる
※プロンプトは公式を参照 ( https://api.python.langchain.com/en/latest/agents/langchain.agents.react.agent.create_react_agent.html )

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate

# ツール定義
@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression."""
    return str(eval(expression))

tools = [calculator]

# ReAct 用プロンプト
prompt = PromptTemplate.from_template(
    """
あなたはタスクを解決するために以下のツールを使うことができます。

利用可能なツール:
{tools}

使用できるツール名一覧: {tool_names}

---
次のフォーマットに従ってください：

Question: ユーザーからの質問
Thought: あなたの思考
Action: 使用するツール名（{tool_names} のいずれか）
Action Input: ツールへの入力
Observation: ツールからの返答
...（必要なだけ Thought → Action → Observation を繰り返す）
Final Answer: 最終的な回答
---

Question: {input}
Thought:{agent_scratchpad}
"""
)

# LLM
llm = ChatOpenAI(model="gpt-4o-mini")

# ReAct Agent 作成 ---
agent = create_react_agent(
    llm=llm,
    tools=tools,
    prompt=prompt,
)

# Executor
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True
)

# 実行
result = executor.invoke({"input": "What is 3 * (2 + 4)?"})
print(result["output"])

プロンプトには create_react_agent が内部的に必須とするプロンプト変数（tools, tool_names, agent_scratchpad）が含まれていないとエラーになる。
※これらの中身は自動挿入される為、input のようにパラメータとして個別に指定する必要はない。

Prompt missing required variables: {'agent_scratchpad', 'tools', 'tool_names'}

説明

create_tool_calling_agent の利用 †

ReAct(create_react_agent) は LLM が文字列で JSON を手打ちする為、壊れやすい(※1)。
そこで、JSON形式でツール呼び出しが出来る create_tool_calling_agent(※2) を利用したサンプルを記載する。
※1 ... 特に単一文字列以外を引数に取る関数を利用する場合にコケるケースが多い（複数の引数や数値などを引数に取る関数の場合）
※2 ... OpenAI の “function calling / tool calling” を使った LangChain の最新の最も安定したエージェント作成方法。

create_tool_calling_agent
https://api.python.langchain.com/en/latest/langchain/agents/langchain.agents.tool_calling_agent.base.create_tool_calling_agent.html

• LLM（gpt-4o / gpt-4o-mini 等）がJSON 形式でツールを呼び出す
• ReAct のように文字列生成で壊れることがない
• 安定して Tools を使った推論ができる
• LangChain v0.2 以降の「標準的なエージェント方式」

コードサンプル

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

@tool
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

tools = [add]
llm = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたはツールを使って問題を解決するアシスタントです。"),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),  # ← これ重要
])

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

result = executor.invoke({"input": "3 と 5 を足してください"})
print(result)

結果

{'input': '3 と 5 を足してください', 'output': '3 と 5 を足すと、8 になります。'}

MCPサーバを利用する †

LangChainでMCPサーバを利用する を参照