Chatbot

ChatBot Skill

概要

マルチターン対話を管理する統一 Skill。会話履歴、コンテキスト維持、Agent/RAG 連携をサポート。

機能

クイックスタート

基本的な使い方

from agentflow.skills.chatbot import ChatBotSkill, ChatBotConfig

# 初期化（最小設定）
chatbot = ChatBotSkill()

# セッション作成
session = chatbot.create_session()

# 対話
response = await chatbot.chat(session.id, "こんにちは！")
print(response)  # "こんにちは！何かお手伝いできますか？"

response = await chatbot.chat(session.id, "天気について教えて")
print(response)  # "申し訳ありませんが、現在の天気情報は..."

カスタム設定

from agentflow.llm.llm_client import LLMConfig

# LLM 設定
llm_config = LLMConfig(
    model="gpt-4o",
    temperature=0.7,
    max_tokens=2000,
)

# ChatBot 設定
bot_config = ChatBotConfig(
    system_prompt="あなたは親切なカスタマーサポートです。",
    max_history=20,
    enable_rag=False,
)

chatbot = ChatBotSkill(llm_config=llm_config, config=bot_config)

RAG 連携

from agentflow.skills.rag import RAGSkill

# RAG Skill を先に初期化
rag = RAGSkill()
await rag.start()

# ドキュメント追加
await rag.add_document("製品マニュアルの内容...", topic="manual")
await rag.add_document("FAQ の内容...", topic="faq")

# RAG 連携 ChatBot
chatbot = ChatBotSkill(
    config=ChatBotConfig(enable_rag=True),
    rag_skill=rag,
)

session = chatbot.create_session()
response = await chatbot.chat(session.id, "製品の使い方を教えて")
# → RAG から関連情報を取得して回答

セッション管理

セッション作成

# デフォルトセッション
session = chatbot.create_session()

# カスタム設定セッション
session = chatbot.create_session(
    system_prompt="あなたは技術サポートです。",
    metadata={"user_id": "user123", "department": "engineering"},
)

print(f"セッション ID: {session.id}")

セッション取得

# ID でセッション取得
session = chatbot.get_session("session_xxx")

# 全セッション一覧
sessions = chatbot.list_sessions()
for s in sessions:
    print(f"{s['id']}: {s['message_count']} messages")

セッションクリア

# 特定セッションをクリア
chatbot.clear_session("session_xxx")

# 全セッションをクリア
chatbot.clear_all_sessions()

会話履歴

履歴取得

# セッションの履歴を取得
messages = chatbot.get_history(session.id)
for msg in messages:
    print(f"[{msg.role}] {msg.content}")

履歴のエクスポート

# JSON 形式でエクスポート
history_json = chatbot.export_history(session.id, format="json")

# Markdown 形式でエクスポート
history_md = chatbot.export_history(session.id, format="markdown")

Agent 連携

Coordinator 連携

from agentflow.patterns.coordinator import AgentCoordinator

# Coordinator を設定
coordinator = AgentCoordinator(agents=[Agent1(), Agent2()])

chatbot = ChatBotSkill(
    config=ChatBotConfig(enable_agent=True),
    coordinator=coordinator,
)

# Agent を呼び出す対話
response = await chatbot.chat(
    session.id, 
    "データを分析して",
    invoke_agent=True,  # Agent 呼び出しを有効化
)

ストリーミング

# ストリーミング応答
async for chunk in chatbot.chat_stream(session.id, "長い説明をして"):
    print(chunk, end="", flush=True)
print()  # 改行

FastAPI 統合

from fastapi import FastAPI, WebSocket
from agentflow.skills.chatbot import ChatBotSkill

app = FastAPI()
chatbot = ChatBotSkill()

@app.post("/api/chat")
async def chat(session_id: str, message: str):
    response = await chatbot.chat(session_id, message)
    return {"response": response}

@app.websocket("/ws/chat/{session_id}")
async def websocket_chat(websocket: WebSocket, session_id: str):
    await websocket.accept()
    while True:
        message = await websocket.receive_text()
        async for chunk in chatbot.chat_stream(session_id, message):
            await websocket.send_text(chunk)

Agent 統合例

from agentflow import agent, AgentClient

@agent
class CustomerSupportAgent:
    """カスタマーサポート Agent"""
    
    system_prompt = "あなたは親切なカスタマーサポートです。"
    
    def __init__(self):
        self.chatbot = ChatBotSkill()
    
    async def run(self, input_data: dict) -> dict:
        session_id = input_data.get("session_id") or self.chatbot.create_session().id
        message = input_data.get("message", "")
        
        response = await self.chatbot.chat(session_id, message)
        return {
            "session_id": session_id,
            "response": response,
        }

設定オプション

ベストプラクティス

1. セッション管理

# ユーザーごとにセッション管理
user_sessions: dict[str, str] = {}

def get_or_create_session(user_id: str) -> str:
    if user_id not in user_sessions:
        session = chatbot.create_session(metadata={"user_id": user_id})
        user_sessions[user_id] = session.id
    return user_sessions[user_id]

2. エラーハンドリング

from agentflow.skills.chatbot import ChatBotError, SessionNotFoundError

try:
    response = await chatbot.chat(session_id, message)
except SessionNotFoundError:
    # セッションが見つからない場合は新規作成
    session = chatbot.create_session()
    response = await chatbot.chat(session.id, message)
except ChatBotError as e:
    logger.error(f"ChatBot エラー: {e}")
    response = "申し訳ありません、エラーが発生しました。"

3. メモリ管理

# 定期的に古いセッションをクリア
import asyncio

async def cleanup_sessions():
    while True:
        await asyncio.sleep(3600)  # 1時間ごと
        chatbot.cleanup_old_sessions(max_age_hours=24)