Langchain

概念

LangChain は、大規模言語モデル（LLM）で動くアプリケーションを構築するためのオープンソースフレームワークです。LLM アプリの全ライフサイクルをカバーします。

• 開発：LangChain のオープンソースの構成ブロック、コンポーネント、サードパーティ統合を使ってアプリケーションを構築します。LangGraph を使うと、一流のストリーミング処理と人間参加型サポートを持つステートフル Agent を構築できます。
• 本番化：LangSmith を使い、chain を検査、監視、評価します。これにより継続的に最適化し、自信を持ってデプロイできます。
• デプロイ：LangGraph Cloud を使い、LangGraph アプリケーションを本番対応の API とアシスタントへ変換します。

実際のプロジェクトでは、通常は開発能力だけを使います。本番化とデプロイは必要に応じて選びます。

Langchain フレームワークは次のオープンソースライブラリで構成されます。

ライブラリ名	役割
langchain-core	基礎抽象層です。Runnable、LCEL などのコアインターフェースを定義します
langchain-community	コミュニティが保守するサードパーティツール統合です（例：ベクトルデータベース、ドキュメントローダー）
langchain	高レベルのアプリケーションコンポーネントです。Chains、Agents、Retrieval など
langchain-openai / langchain-anthropic など	軽量な LLM 統合パッケージです（langchain-core のみに依存）
LangGraph	ステートフル、マルチステップ、人間参加型の Agent を構築し、グラフ構造でプロセスをモデル化します
LangServe	LangChain chain を RESTful API としてデプロイします
LangSmith	LLM アプリをデバッグ、テスト、監視するための開発者プラットフォームです

LCEL

LCEL の正式名称は LangChain Expression Language です。主な用途は | 演算子で LangChain アプリケーションの各コンポーネントを接続することです。

特徴：組み合わせ可能、並列可能、ストリーミング可能、フォールバック可能、監視可能です。コードを変更せずに本番環境で使えます。

LCEL の各種構文

開発時にインストールが必要なライブラリ：

langchain
langchain-openai
langchain-community
LangGraph

基本呼び出し（非 LCEL）

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

llm = ChatOpenAI(
    model="GLM-4.5-Flash",
    temperature=1,
    openai_api_key='',
    openai_api_base=''
)
prompt = [
        SystemMessage('次の文を日本語へ翻訳してください'), # 内部の波括弧はパラメータ定義を表す
		HumanMessage('今日の天気はどうですか？')
	]
parser = StrOutputParser() # 応答オブジェクトを文字列へ解析

# 一つ目の書き方
res = llm.invoke(prompt)
parser.invoke(res)
print(res)
print(parser)

LCEL で大規模モデルを呼び出す（よく使う）

# Prompt テンプレート
prompt = ChatPromptTemplate.from_messages(
	[
        ('system', '次の文を{language}へ翻訳してください'), # 内部の波括弧はパラメータ定義を表す
        ('user', '{user_text}')
	]
)
parser = StrOutputParser() # 応答オブジェクトを文字列へ解析

# LCEL
chain = prompt | llm | parser
res = chain.invoke({'language': '日本語', 'user_text': 'ご飯を食べましたか？'})

print(res)

1. Runnable ノード

すべての LCEL コンポーネントは Runnable を継承しており、.invoke() で呼び出せます。

from langchain_core.runnables import RunnableLambda

def add_ten(x: int) -> int:
    return x + 10

node = RunnableLambda(add_ten)
print(node.invoke(5))  # 出力: 15

2. 複数の実行モード

• バッチ呼び出し（Batch）

results = node.batch([1, 2, 3])  # [11, 12, 13]

• ストリーミング出力

def word_stream(prompt: str):
    for word in prompt.split():
        yield word

stream_node = RunnableLambda(word_stream)
for chunk in stream_node.stream("Hello world from LCEL"):
    print(chunk)  # 単語ごとに出力

3. 組み合わせと並列

• chain 実行（Sequential）

double = RunnableLambda(lambda x: x * 2)
add_five = RunnableLambda(lambda x: x + 5)

# ノードを chain として組み合わせる。「|」を使う
chain = double | add_five
print(chain.invoke(3))  # (3*2)+5 = 11

• 並列実行（Parallel）

from langchain_core.runnables import RunnableParallel

# 並列
parallel_chain = RunnableParallel(
    result1=double,
    result2=add_five
)
print(parallel_chain.invoke(4))  # {'result1': 8, 'result2': 9}

# 可視化フローチャートを表示（graphviz のインストールが必要）
parallel_chain.get_graph().print_ascii()

4. 中間データ処理：RunnablePassthrough

chain 内で入力データを保持、拡張、またはフィルタリングするために使います。

from langchain_core.runnables import RunnablePassthrough

# ステップ1：入力を辞書に包む
wrap = RunnableLambda(lambda x: {"value": x})

# ステップ2：辞書に基づいて新しいフィールドを計算
compute = RunnableLambda(lambda d: d["value"] * 2)

# 組み合わせ：元データを保持 + 新フィールドを追加 + 新フィールドだけ出力
chain = (
    wrap 
    | RunnablePassthrough.assign(doubled=compute) 
    | RunnablePassthrough.pick(["doubled"])
)

print(chain.invoke(5))  # {'doubled': 10}

よく使うメソッド：

• .assign(key=runnable)：新しいキーを追加します
• .pick(keys)：指定キーを選びます
• RunnablePassthrough()：入力をそのまま通します

5. フォールトトレランス：Fallbacks

主コンポーネントが失敗した時、自動で代替案へ切り替えます。

def risky_func(x):
    if isinstance(x, str):
        raise ValueError("Expected int")
    return x + 10

safe_func = RunnableLambda(lambda x: int(x) + 5)

main = RunnableLambda(risky_func)
fallback_chain = main.with_fallbacks([safe_func])

print(fallback_chain.invoke("3"))  # 主処理失敗 → fallback 実行 → 8 を出力

6. Retry

失敗した操作を自動で再試行します。ネットワークリクエストや不安定なコンポーネントでよく使います。

import time

counter = 0
def flaky_divide(x):
    global counter
    counter += 1
    print(f"Attempt {counter}")
    if counter < 3:
        raise Exception("Simulated failure")
    return x / 2

retriable = RunnableLambda(flaky_divide).with_retry(stop_after_attempt=3)
print(retriable.invoke(10))  # 3回目で成功し、5.0 を出力

7. 条件分岐（Conditional Chaining）

中間結果に応じて、後続ロジックを動的に選びます。

def route_by_value(x):
    if x > 10:
        return RunnableLambda(lambda _: "Large number")
    else:
        return x  # 値を直接返す（Runnable ではない）

chain = (
    RunnableLambda(lambda x: x + 5)  # 先に5を足す
    | RunnableLambda(route_by_value)
)

print(chain.invoke(3))   # 3+5=8 ≤10 → 8 を出力
print(chain.invoke(10))  # 10+5=15 >10 → "Large number" を出力

注意：条件関数は Runnable または直接出力できる値を返す必要があります。

8. ライフサイクル監視（Listeners）

コンポーネント実行の前後にコールバックを挿入し、ログ、監視、計時などに使います。

import time
from langchain_core.tracers import Run

def on_start(run: Run):
    print(f"Started at: {run.start_time}")

def on_end(run: Run):
    print(f"Ended at: {run.end_time}, Duration: {run.end_time - run.start_time}")

slow_task = RunnableLambda(lambda x: time.sleep(x) or x * 2)
monitored_chain = slow_task.with_listeners(on_start=on_start, on_end=on_end)

monitored_chain.invoke(1)  # 開始/終了時間を表示

コンテキスト付き対話（履歴の永続化）

コンテキスト履歴はローカルに保存し、ディスク、つまりデータベースに置く必要があります。

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import SQLChatMessageHistory

# セッション履歴を取得する関数を定義
def get_session_history(session_id: str):
    return SQLChatMessageHistory(session_id, connection="sqlite:///chat.db")

# memory 付き chain を構築
prompt_with_history = ChatPromptTemplate.from_messages([
    ("system", "あなたはユーモアのあるプログラマーです"),
    MessagesPlaceholder(variable_name="history"),
    ("user", "{input}")
])

base_chain = prompt_with_history | llm | StrOutputParser()

runnable = RunnableWithMessageHistory(
    base_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 同じセッション ID の複数回呼び出しはコンテキストを共有する
res1 = runnable.invoke(
    {"input": "中国には直轄市がいくつありますか？"},
    config={"configurable": {"session_id": "user123"}}
)
res2 = runnable.invoke(
    {"input": "その中で一番大きい都市はどれですか？"},
    config={"configurable": {"session_id": "user123"}}
)