LangChain スマートカスタマーサービスボット

# ファイルパス: check_install.py
import os
from dotenv import load_dotenv

load_dotenv()

# 依存パッケージが正常にインポートできるか確認
import langchain
import langchain_deepseek
import langchain_openai
import langchain_chroma
import chromadb
import langgraph

print(f"langchain バージョン: {langchain.__version__}")

# キーが設定されているか確認
assert os.getenv("DEEPSEEK_API_KEY"), "DEEPSEEK_API_KEY が検出されませんでした。.env ファイルを確認してください"
# assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY が検出されませんでした。.env ファイルを確認してください"
assert os.getenv("DASHSCOPE_API_KEY"), "DASHSCOPE_API_KEY が検出されませんでした。.env ファイルを確認してください"

print("環境設定が成功しました～カスタマーサービスボットの作成を始められます！")

# ファイルパス: customer_service_bot.py
# 依存パッケージのインストール、.env の設定は前の節「環境構築」を参照
from dotenv import load_dotenv
load_dotenv()

import os
import sqlite3
from typing import Annotated
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import before_model, after_model
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, AIMessage
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt, Command


# ========== 1. ナレッジベースの準備 ==========

knowledge_base = [
    "サンプルは2013年に設立され、中国国内でトップクラスの無料プログラミング学習プラットフォームです。",
    "このプラットフォームは、Python、Java、HTML、CSS、JavaScript などを含む 300 以上のチュートリアルを提供しています。",
    "Python3 基礎チュートリアルは全30章で、累計学習者数は500万人を超えています。コースは完全に無料です。",
    "VIP 会員費は月額 ¥99、年額 ¥799 で、ビデオコースと1対1の質疑応答サービスが含まれます。",
    "返金ポリシー：購入後7日以内かつ3レッスン以内であれば全額返金できます。",
    "プラットフォームはオンラインプログラミング環境をサポートしており、ソフトウェアをインストールせずにコードを記述・実行できます。",
    "カスタマーサービスの受付時間：月曜日から金曜日 9:00〜18:00、土日 10:00〜16:00。",
]

# アリババクラウド百炼（DashScope）の通義千問（Qwen）Embedding サービスを使用
# 百炼の Embedding API は OpenAI API 仕様と互換性があるため、langchain-openai の
# OpenAIEmbeddings を直接使い、base_url を百炼の互換エンドポイントに指定するだけでよいです。
# langchain-community / dashscope をインストールする必要はありません（このパッケージはメンテナンスを停止しています）。
# text-embedding-v4 は現在推奨されている汎用ベクトルモデルで、デフォルトで 1024 次元のベクトルを出力します。
#
# 欠かせない重要なパラメータが2つあります：
# - check_embedding_ctx_length=False：OpenAIEmbeddings はデフォルトで tiktoken を使用します
テキストを事前に token ID 配列にエンコードしてから送信します（OpenAI 公式 API はこの形式を認識します）。
ただし、百炼の互換 API は元の文字列のみを受け付けるため、このオプションをオフにしないと、
"contents is neither str nor list of str" というエラーが発生します。
# - chunk_size=10：百炼 Embedding API は1回のリクエストで最大10件のテキストしか受け付けません
# OpenAIEmbeddings はデフォルトで一度に1000件をまとめて送信するため、ナレッジベースが少し大きいと上限を超えてエラーになります。
embeddings = OpenAIEmbeddings(
    model="text-embedding-v4",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    check_embedding_ctx_length=False,
    chunk_size=10,
)
chunks = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=30
                                         ).create_documents(knowledge_base)
vector_store = Chroma.from_documents(chunks, embeddings)
retriever = vector_store.as_retriever(search_kwargs={"k": 3})


# ========== 2. ツールの定義 ==========

@tool
def search_kb(query: str) -> str:
    """サンプルのナレッジベースを検索し、プラットフォーム、コース、ポリシーなどの公式情報を取得します。

    Args:
query: 検索する質問またはキーワード
    """
    docs = retriever.invoke(query)
    if not docs:
        return "関連情報が見つかりませんでした。オペレーターへの転送をお勧めします。"
    return "\n".join(f"- {doc.page_content}" for doc in docs)


# 模擬注文データベース
orders_db = {
    "ORD-2024-001": {"user": "小明", "item": "VIP 年間会員",
                      "amount": 799, "status": "完了", "date": "2024-01-15"},
    "ORD-2024-002": {"user": "小明", "item": "Python 実践コース",
                      "amount": 199, "status": "配送中", "date": "2024-03-20"},
}


@tool
def query_order(order_id: str) -> str:
    """注文番号に基づいて注文のステータスと詳細を照会します。

    Args:
order_id: 注文番号（例: ORD-2024-001）
    """
    order = orders_db.get(order_id.upper())
    if not order:
        return f"注文 {order_id} が見つかりません。注文番号が正しいか確認してください。"
    return (f"注文 {order_id}：{order['item']} | "
            f"金額 ¥{order['amount']} | "
            f"ステータス {order['status']} | "
            f"日付 {order['date']}")


@tool
def transfer_to_human(reason: str) -> str:
    """ユーザーをオペレーターに転送します。

    Args:
reason: 転送理由
    """
    approval = interrupt({
        "action": "transfer_to_human",
        "reason": reason,
        "message": f"ユーザーがオペレーターへの転送を要求しています。理由：{reason}。転送しますか？"
    })
    if approval.get("confirmed"):
        return (f"オペレーターに転送しました。予想待ち時間は {approval.get('wait_time', 3)} 分です。"
                fチケット番号：TK-{approval.get('ticket_id', 'N/A')})
    return お取り次ぎはキャンセルされました。引き続き対応させていただきます。


# ========== 3. Middleware の定義 ==========

@before_model
def content_guard(state, runtime):
    """ユーザー入力の不適切な内容をフィルタリングする"""
    last_msg = state["messages"][-1] if state.get("messages") else None
    if not last_msg:
        return None
    content = str(getattr(last_msg, 'content', ''))
    blocked = [黄X, X博, 違法]
    for word in blocked:
        if word in content:
            return {
                "jump_to": "end",
                "messages": [HumanMessage(content=申し訳ありませんが、このリクエストは処理できません。)]
            }
    return None


@after_model
def auto_signature(state, runtime):
    """カスタマーサポートの署名を自動的に追加する"""
    msgs = state.get("messages", [])
    if not msgs:
        return None
    last = msgs[-1]
    if last.type == "ai" and last.content and not (
        hasattr(last, 'tool_calls') and last.tool_calls
    ):
        # ポイント：last.id を再利用して、add_messages reducer がそのメッセージをその場で置き換えるようにする。
        # 履歴に新しいメッセージとして追加するのではなく（そうしないと履歴がどんどん肥大化し、
        # 毎ターン「署名なし版」と「署名付き版」が1件ずつ増えてしまう）
        return {"messages": [AIMessage(
            id=last.id,
            content=last.content
            + "\n\n---\nサンプル カスタマーセンター | 営業時間 9:00-18:00"
        )]}
    return None


# ========== 4. Agent の作成 ==========

# SqliteSaver.from_conn_string() はコンテキストマネージャーを返すため、「使い終わったら閉じる」だけの
# 使い捨てスクリプトにのみ適している。カスタマーサービスボットは複数の chat() 呼び出し間で同じデータベース接続を維持する必要があるため、
# ここでは自分で接続を確立してから SqliteSaver コンストラクタに渡している。
# check_same_thread=False は、Web フレームワークが通常、同じ接続をスレッド間で呼び出すためである。
conn = sqlite3.connect("customer_service.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)

agent = create_agent(
    model=model,
    tools=[search_kb, query_order, transfer_to_human],
    middleware=[content_guard, auto_signature],
    checkpointer=checkpointer,
    system_prompt="""あなたはサンプル のスマートカスタマーサービス「小菜」です。

## あなたの役割
1. すべてのユーザーを熱意を持って対応し、「您」という敬称で呼ぶ
2. プラットフォーム情報、コース内容、ポリシーなどの問題については、search_kb を使って検索する
3. 注文照会については、query_order ツールを使用する
4. 解決できない問題に遭遇した場合は、transfer_to_human を使用して人工オペレーターに転送する

## 行動規範
- 回答は簡潔に、毎回 2〜3 文で
- 分からない場合はナレッジベースを検索し、見つからない場合は正直に伝える
- 友好的で親しみやすい口調を保つ""",
)


# ========== 5. 対話インターフェース ==========

def chat(thread_id: str, message: str) -> str:
    """ユーザーメッセージを処理して返信を返す"""
    config = {"configurable": {"thread_id": thread_id}}

    # Agent を実行
    result = agent.invoke(
        {"messages": [HumanMessage(content=message)]},
        config=config,
    )

    # 転送が必要かどうかを確認する（HITL）
    state = agent.get_state(config)
    if state.tasks and state.tasks[0].interrupts:
        interrupt_info = state.tasks[0].interrupts[0].value
        return f"[承認が必要] {interrupt_info.get('message', '')}"

    return result["messages"][-1].content


def resume_transfer(thread_id: str, confirmed: bool,
                     wait_time: int = 3, ticket_id: str = "0001") -> str:
    """人工オペレーターがバックグラウンドで承認した後、interrupt() によって中断された転送フローを再開します。

transfer_to_human ツール内で待機している approval データに対応します。
    """
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        Command(resume={
            "confirmed": confirmed,
            "wait_time": wait_time,
            "ticket_id": ticket_id,
        }),
        config=config,
    )
    return result["messages"][-1].content


# ========== 6. テスト ==========

if __name__ == "__main__":
    user_id = "user_xiaoming"

    print("=== テスト 1：ナレッジベース検索 ===")
    print(chat(user_id, "Python3 チュートリアルは何章ありますか？"))
    print()

    print("=== テスト 2：注文照会 ===")
    print(chat(user_id, "私の注文 ORD-2024-001 のステータスは何ですか？"))
    print()

    print("=== テスト 3：VIP 相談 ===")
    print(chat(user_id, "VIP 会員はいくらですか？"))
    print()

    print("=== テスト 4：メモリのテスト ===")
    print(chat(user_id, "さっき私はどんな質問をしましたか？"))
    print()

    print("=== テスト 5：人工オペレーターへの転送（HITL） ===")
    print(chat(user_id, "苦情があります。人工オペレーターに転送してください"))       # interrupt をトリガーし、承認を待つ
    print(resume_transfer(user_id, confirmed=True,
                           wait_time=5, ticket_id="8823"))  # バックグラウンドで確認後、再開する

    conn.close()