【LangChain】LLMの文脈消失問題を解決するセッション管理法

ステップ5: 完全なチャットボットを実装する

以上の要素を組み合わせた完全な実装がこちらです。

from langchain_core.runnables import RunnablePassthrough

class PersistentContextChatbot:
    def __init__(self, llm):
        self.llm = llm
        self.session_histories = {}
        self.session_instructions = {}
    
    def get_history(self, session_id: str):
        if session_id not in self.session_histories:
            self.session_histories[session_id] = ChatMessageHistory()
        return self.session_histories[session_id]
    
    def update_instructions(self, session_id: str, instruction: str):
        if session_id not in self.session_instructions:
            self.session_instructions[session_id] = []
        if instruction not in self.session_instructions[session_id]:
            self.session_instructions[session_id].append(instruction)
    
    def get_instructions(self, session_id: str) -> str:
        if session_id not in self.session_instructions:
            return "常に准确かつ简洁に回答してください。"
        return " ".join([f"[{i+1}] {inst}" 
                        for i, inst in enumerate(self.session_instructions[session_id])])
    
    def build_prompt(self, session_id: str, user_input: str):
        history = self.get_history(session_id)
        instructions = self.get_instructions(session_id)
        
        context_system = f"""
你是可靠的AI助手。
重要な指示: {instructions}

会話履歴を考慮して回答してください。
"""
        messages = [("system", context_system)]
        
        for msg in history.messages:
            messages.append((msg.type, msg.content))
        
        messages.append(("human", user_input))
        
        return ChatPromptTemplate.from_messages(messages)
    
    def chat(self, session_id: str, user_input: str) -> str:
        # 指示の検出と保存
        if "JSON" in user_input.upper() or "形式" in user_input:
            self.update_instructions(session_id, user_input)
        
        # プロンプトを構築
        prompt = self.build_prompt(session_id, user_input)
        
        # LLMを呼び出し
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({})
        
        # 履歴に保存
        history = self.get_history(session_id)
        history.add_user_message(user_input)
        history.add_ai_message(response)
        
        return response

# 使用例
chatbot = PersistentContextChatbot(llm)
session_id = "user_123"

# 重要な指示を与える
print(chatbot.chat(session_id, "今後は常にJSON形式で回答してください"))
# 通常の会話
print(chatbot.chat(session_id, "私の名前は田中です"))
# 数ターン後
print(chatbot.chat(session_id, "私の名前は何ですか？"))  # JSON形式で回答される

from langchain.memory import ConversationSummaryMemory

class SummarizingChatbot:
    def __init__(self, llm):
        self.llm = llm
        self.sessions = {}
    
    def get_memory(self, session_id: str):
        if session_id not in self.sessions:
            self.sessions[session_id] = ConversationSummaryMemory(
                llm=self.llm,
                memory_key="chat_history",
                return_messages=True
            )
        return self.sessions[session_id]
    
    def chat(self, session_id: str, user_input: str) -> str:
        memory = self.get_memory(session_id)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "あなたは一貫した помощник です。重要な指示: {instructions}"),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{input}")
        ])
        
        # 重要な指示を取得（別途管理が必要）
        instructions = get_important_instructions(session_id)
        
        chain = RunnablePassthrough.assign(
            chat_history=memory.load_memory_variables({})["chat_history"],
            instructions=lambda x: instructions
        ) | prompt | self.llm | StrOutputParser()
        
        response = chain.invoke({"input": user_input})
        
        # メモリに保存
        memory.save_context(
            {"input": user_input},
            {"output": response}
        )
        
        return response