ユーザーの質問
    ↓
① ベクトル検索で関連文書を引っ張ってくる
    ↓
② LLMに「こういう資料があるよ、これを参考に答えてね」と渡す
    ↓
③ LLMが根拠ある回答を生成する

# 動作確認済み：Python 3.11 / langchain 0.3.x / chromadb 0.5.x
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA

class BasicRAG:
    def __init__(self, documents_path: str, chunk_size: int = 500):
        # ドキュメント読み込み
        loader = TextLoader(documents_path, encoding="utf-8")
        documents = loader.load()

        # チャンク分割（chunk_sizeは用途に応じて調整）
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=50  # 文脈を失わないよう重複を持たせる
        )
        chunks = splitter.split_documents(documents)

        # ベクトルDBに保存（タイムアウトを明示的に設定）
        embeddings = OpenAIEmbeddings(model="text-embedding-3-small", timeout=60)
        self.vectorstore = Chroma.from_documents(chunks, embeddings)

        # QAチェーン構築（タイムアウトを明示的に設定）
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=ChatOpenAI(model="gpt-4o", temperature=0, timeout=60),
            retriever=self.vectorstore.as_retriever(search_kwargs={"k": 3}),
            return_source_documents=True
        )

    def query(self, question: str) -> dict:
        """質問に対して関連文書を検索し、根拠付きで回答を生成"""
        result = self.qa_chain.invoke({"query": question})
        return {
            "answer": result["result"],
            "sources": [doc.metadata for doc in result["source_documents"]]
        }

# 使用例
if __name__ == "__main__":
    rag = BasicRAG("./company_docs.txt")
    response = rag.query("有給休暇の申請方法は？")
    print(response["answer"])

# 動作確認済み：BM25（キーワード検索）＋ベクトル検索の組み合わせ
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

def build_hybrid_retriever(documents: list, weights: list = None):
    """
    BM25（キーワード検索）とベクトル検索を組み合わせたハイブリッドリトリーバー
    weights: [bm25_weight, vector_weight]
    """
    if weights is None:
        weights = [0.5, 0.5]

    # BM25リトリーバー（キーワードベース）
    bm25_retriever = BM25Retriever.from_documents(documents)
    bm25_retriever.k = 3

    # ベクトルリトリーバー（意味ベース）（タイムアウトを明示的に設定）
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small", timeout=60)
    vectorstore = Chroma.from_documents(documents, embeddings)
    vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

    # 両者をアンサンブルで統合
    ensemble_retriever = EnsembleRetriever(
        retrievers=[bm25_retriever, vector_retriever],
        weights=weights
    )
    return ensemble_retriever

# 動作確認済み：Agentic RAG実装例（LangGraph使用）
import json
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from typing import TypedDict, List

class AgentState(TypedDict):
    question: str
    sub_queries: List[str]
    retrieved_docs: List[str]
    answer: str
    iterations: int

def decompose_query(state: AgentState) -> AgentState:
    """複雑な質問を検索可能なサブクエリに分解"""
    llm = ChatOpenAI(model="gpt-4o", temperature=0, timeout=60)
    prompt = f"""以下の質問を、それぞれ独立して検索できる2〜3つのサブクエリに分解してください。
質問: {state['question']}
JSON形式で返してください: {{"sub_queries": ["クエリ1", "クエリ2"]}}"""

    response = llm.invoke([HumanMessage(content=prompt)])
    result = json.loads(response.content)
    # iterationsを確実にインクリメントして無限ループを防止
    return {
        **state,
        "sub_queries": result["sub_queries"],
        "iterations": state.get("iterations", 0) + 1,
    }

def retrieve_docs(state: AgentState) -> AgentState:
    """サブクエリに基づいてドキュメントを取得"""
    retrieved = [f"Retrieved doc for: {q}" for q in state.get("sub_queries", [])]
    return {
        **state,
        "retrieved_docs": state.get("retrieved_docs", []) + retrieved,
    }

def evaluate_docs(state: AgentState) -> AgentState:
    """取得した情報の十分性を評価（状態をそのまま返し、条件分岐はエッジで処理）"""
    return {**state}

def generate_answer(state: AgentState) -> AgentState:
    """収集したドキュメントを元に最終回答を生成"""
    llm = ChatOpenAI(model="gpt-4o", temperature=0, timeout=60)
    context = "\n".join(state.get("retrieved_docs", []))
    prompt = f"""以下のコンテキストを元に質問に答えてください。
質問: {state['question']}
コンテキスト: {context}"""
    response = llm.invoke([HumanMessage(content=prompt)])
    return {**state, "answer": response.content}

def check_sufficiency(state: AgentState) -> str:
    """取得した情報が十分かどうかを判断して次のノードを決定"""
    if len(state.get("retrieved_docs", [])) < 2 and state.get("iterations", 0) < 3:
        return "retrieve_more"
    return "generate_answer"

# グラフ構築（全ノード・エッジを明示的に定義）
workflow = StateGraph(AgentState)

workflow.add_node("decompose", decompose_query)
workflow.add_node("retrieve", retrieve_docs)
workflow.add_node("evaluate", evaluate_docs)
workflow.add_node("answer", generate_answer)

workflow.set_entry_point("decompose")
workflow.add_edge("decompose", "retrieve")
workflow.add_edge("retrieve", "evaluate")

workflow.add_conditional_edges("evaluate", check_sufficiency, {
    "retrieve_more": "decompose",
    "generate_answer": "answer",
})
workflow.add_edge("answer", END)

app = workflow.compile()

# 実行例
initial_state = AgentState(
    question="LLMアプリの設計パターンにはどんなものがありますか？",
    sub_queries=[],
    retrieved_docs=[],
    answer="",
    iterations=0,
)
result = app.invoke(initial_state)
print(result["answer"])

# 動作確認済み：Python 3.11 / langchain 0.3.x
from langchain_openai import ChatOpenAI
from langchain.memory import (
    ConversationSummaryMemory,
    ConversationBufferMemory,
    ConversationBufferWindowMemory,
)
from langchain.chains import ConversationChain

def build_memory_agent(memory_type: str = "summary", window_k: int = 5):
    """
    メモリタイプに応じたエージェントを構築
    memory_type: "summary" | "window" | "buffer"
    """
    llm = ChatOpenAI(model="gpt-4o", temperature=0.7, timeout=60)

    if memory_type == "summary":
        # 長期会話向け：要約して圧縮（トークンコストを抑える）
        memory = ConversationSummaryMemory(
            llm=llm,
            return_messages=True,
            max_token_limit=2000
        )
    elif memory_type == "window":
        # バランス重視：直近k件のみ保持
        memory = ConversationBufferWindowMemory(
            k=window_k,
            return_messages=True
        )
    else:
        # 短い会話・精度重視：全履歴を保持
        memory = ConversationBufferMemory(return_messages=True)

    chain = ConversationChain(llm=llm, memory=memory, verbose=False)
    return chain

# 使用例：サマリーメモリを使った長期会話エージェント
agent = build_memory_agent(memory_type="summary")

for message in [
    "私はPythonエンジニアです。RAGシステムを開発しています。",
    "チャンクサイズはどうすればいいですか？",
    "先ほど話した私の職業に合わせてアドバイスをください。",  # メモリテスト
]:
    response = agent.predict(input=message)
    print(f"Q: {message}\nA: {response}\n")

# 動作確認済み：ツール付きエージェント実装
from datetime import datetime
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

@tool
def get_current_time() -> str:
    """現在の日時を取得する"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def calculate(expression: str) -> str:
    """数式を計算する。例: '2 + 3 * 4'"""
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as e:
        return f"計算エラー: {e}"

@tool
def search_web(query: str) -> str:
    """Web検索を実行する（デモ用：実際はTavilySearchResultsを使用）"""
    # 本番では: return TavilySearchResults(max_results=3).run(query)
    return f"検索結果（デモ）: '{query}' に関する情報..."

def build_tool_agent():
    """ツール付きエージェントを構築"""
    llm = ChatOpenAI(model="gpt-4o", temperature=0, timeout=60)
    tools = [get_current_time, calculate, search_web]

    prompt = ChatPromptTemplate.from_messages([
        ("system", "あなたは有能なアシスタントです。必要に応じてツールを使って正確な回答を提供してください。"),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ])

    agent = create_openai_functions_agent(llm, tools, prompt)
    return AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=5)

# 使用例
agent_executor = build_tool_agent()
result = agent_executor.invoke({
    "input": "今何時ですか？また、123 × 456 を計算してください。"
})
print(result["output"])

# 動作確認済み：オーケストレーター＋ワーカーパターン（LangGraph使用）
import json
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from typing import TypedDict, List, Literal

class MultiAgentState(TypedDict):
    task: str
    plan: List[str]
    results: List[str]
    current_step: int
    final_answer: str

def orchestrator(state: MultiAgentState) -> MultiAgentState:
    """タスクを分析してサブタスクのプランを生成する司令塔"""
    llm = ChatOpenAI(model="gpt-4o", temperature=0, timeout=60)
    prompt = f"""以下のタスクを実行するための具体的なサブタスクリストを3〜5個作成してください。
タスク: {state['task']}
JSON形式で: {{"steps": ["ステップ1", "ステップ2", ...]}}"""
    response = llm.invoke([HumanMessage(content=prompt)])
    plan_data = json.loads(response.content)
    return {
        **state,
        "plan": plan_data["steps"],
        "current_step": 0,
        "results": [],
    }

def worker(state: MultiAgentState) -> MultiAgentState:
    """現在のステップを実行するワーカーエージェント"""
    llm = ChatOpenAI(model="gpt-4o", temperature=0.3, timeout=60)
    current_step = state["current_step"]
    step_description = state["plan"][current_step]

    prompt = f"""以下のステップを実行してください。
メインタスク: {state['task']}
現在のステップ: {step_description}
これまでの結果: {state['results']}"""

    response = llm.invoke([HumanMessage(content=prompt)])
    new_results = state["results"] + [f"Step {current_step + 1}: {response.content}"]
    return {
        **state,
        "results": new_results,
        "current_step": current_step + 1,
    }

def synthesizer(state: MultiAgentState) -> MultiAgentState:
    """全ワーカーの結果を統合して最終回答を生成"""
    llm = ChatOpenAI(model="gpt-4o", temperature=0, timeout=60)
    results_text = "\n".join(state["results"])
    prompt = f"""以下の各ステップの結果を統合して、最終的な回答を生成してください。
元のタスク: {state['task']}
各ステップの結果:\n{results_text}"""
    response = llm.invoke([HumanMessage(content=prompt)])
    return {**state, "final_answer": response.content}

def should_continue(state: MultiAgentState) -> Literal["worker", "synthesize"]:
    """全ステップが完了したか確認"""
    if state["current_step"] < len(state["plan"]):
        return "worker"
    return "synthesize"

# グラフ構築
workflow = StateGraph(MultiAgentState)
workflow.add_node("orchestrate", orchestrator)
workflow.add_node("work", worker)
workflow.add_node("synthesize", synthesizer)

workflow.set_entry_point("orchestrate")
workflow.add_edge("orchestrate", "work")
workflow.add_conditional_edges("work", should_continue, {
    "worker": "work",
    "synthesize": "synthesize",
})
workflow.add_edge("synthesize", END)

multi_agent_app = workflow.compile()

# 実行例
result = multi_agent_app.invoke({
    "task": "2026年のLLMアプリ開発トレンドをまとめたブログ記事の構成を作成してください",
    "plan": [],
    "results": [],
    "current_step": 0,
    "final_answer": "",
})
print(result["final_answer"])

# 動作確認済み：パイプラインパターン実装
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from dataclasses import dataclass, field
from typing import List, Callable

@dataclass
class PipelineState:
    raw_content: str
    translated: str = ""
    summarized: str = ""
    report: str = ""
    errors: List[str] = field(default_factory=list)

class AgentPipeline:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0, timeout=60)
        self.steps: List[Callable] = []

    def add_step(self, func: Callable):
        """パイプラインにステップを追加（メソッドチェーン対応）"""
        self.steps.append(func)
        return self

    def run(self, state: PipelineState) -> PipelineState:
        """パイプラインを順番に実行。エラー発生時は以降をスキップ"""
        for step in self.steps:
            try:
                state = step(state, self.llm)
            except Exception as e:
                state.errors.append(f"{step.__name__}: {str(e)}")
                break
        return state

def translate_step(state: PipelineState, llm) -> PipelineState:
    """英語テキストを日本語に翻訳"""
    response = llm.invoke([HumanMessage(
        content=f"以下を日本語に翻訳してください:\n{state.raw_content}"
    )])
    state.translated = response.content
    return state

def summarize_step(state: PipelineState, llm) -> PipelineState:
    """翻訳済みテキストを200文字以内で要約"""
    response = llm.invoke([HumanMessage(
        content=f"以下を200文字以内で要約してください:\n{state.translated}"
    )])
    state.summarized = response.content
    return state

def report_step(state: PipelineState, llm) -> PipelineState:
    """要約からビジネスレポートを生成"""
    response = llm.invoke([HumanMessage(
        content=f"以下の要約を元に、ビジネスレポートを作成してください:\n{state.summarized}"
    )])
    state.report = response.content
    return state

# パイプライン構築と実行
pipeline = (AgentPipeline()
    .add_step(translate_step)
    .add_step(summarize_step)
    .add_step(report_step))

initial = PipelineState(raw_content="LLM applications are growing rapidly in 2026...")
result = pipeline.run(initial)

if result.errors:
    print(f"エラー発生: {result.errors}")
else:
    print(result.report)

LLMアプリを作りたい
    ↓
Q1: 外部の知識・最新情報が必要？
    YES → RAGパターン
        ├─ 固有名詞・専門用語が多い？ → ハイブリッド検索RAG
        ├─ クエリが複雑・多段階調査が必要？ → Agentic RAG
        └─ シンプルなFAQ・文書検索 → ベーシックRAG
    ↓
Q2: 会話の文脈を保持する必要がある？
    YES → メモリ付きエージェント
    ↓
Q3: 外部API・ツールを自律的に使う必要がある？
    YES → ツール使用エージェント
    ↓
Q4: 複数の専門タスクを分業して処理したい？
    YES → マルチエージェント
        ├─ 順序固定の処理 → パイプラインパターン
        └─ 柔軟な分業が必要 → オーケストレーター＋ワーカー