Compound AI Systems - 複合AIシステムの設計パターン

入力 → 単一LLM → 出力

入力 → Retriever（検索） → LLM（推論） → Agent（判断）  
       → Tools（実行） → 統合 → 高精度な出力

# RAG + LLMの基本構成
from langchain.vectorstores import Chroma
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA

# 1. Retriever設定
vectorstore = Chroma.from_documents(documents, embedding_function)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 2. LLM設定
llm = ChatOpenAI(model="gpt-4", temperature=0)

# 3. RAGチェーン構築
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever
)

# 実行
response = qa_chain.run("2025年のAI市場規模は？")

from langchain.retrievers import EnsembleRetriever

# 複数のRetriever
vector_retriever = vectorstore.as_retriever()
bm25_retriever = BM25Retriever.from_documents(documents)
graph_retriever = GraphRetriever(knowledge_graph)

# アンサンブル
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever, graph_retriever],
    weights=[0.4, 0.3, 0.3]
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=ensemble_retriever
)

from langchain.agents import initialize_agent, Tool
from langchain.tools import DuckDuckGoSearchRun, WikipediaQueryRun

# ツール定義
search = DuckDuckGoSearchRun()
wikipedia = WikipediaQueryRun()
calculator = load_tools(["llm-math"], llm=llm)[0]

tools = [
    Tool(name="Search", func=search.run, description="最新情報検索"),
    Tool(name="Wikipedia", func=wikipedia.run, description="百科事典検索"),
    Tool(name="Calculator", func=calculator.run, description="計算実行")
]

# Agentシステム
agent = initialize_agent(
    tools,
    llm,
    agent="zero-shot-react-description",
    verbose=True
)

# 複雑なタスクを自律実行
agent.run("テスラの2025年第1四半期の売上を調べて、前年比の成長率を計算してください")

class CompoundAISystem:
    def __init__(self):
        self.retriever = self._setup_retriever()
        self.llm = self._setup_llm()
        self.agent = self._setup_agent()
        self.tools = self._setup_tools()
    
    def _setup_retriever(self):
        # Retriever最適化
        return HybridRetriever(
            vector_weight=0.6,
            keyword_weight=0.4
        )
    
    def _setup_llm(self):
        # LLM選択（タスク別）
        return ChatOpenAI(model="gpt-4", temperature=0.1)
    
    def _setup_agent(self):
        # Agent設定
        return ReActAgent(tools=self.tools)
    
    def process(self, query):
        # パイプライン実行
        docs = self.retriever.retrieve(query)
        context = self._format_context(docs)
        response = self.agent.run(query, context)
        return response

from langchain.callbacks import LangChainTracer

tracer = LangChainTracer()

# トレーシング有効化
qa_chain.run(
    "質問",
    callbacks=[tracer]
)

# 各コンポーネントの実行時間、コストを分析
tracer.get_run_stats()

class EnterpriseSearchSystem:
    def __init__(self):
        # Multi-source Retrieval
        self.doc_retriever = ChromaDB(collection="documents")
        self.code_retriever = CodeSearchEngine(repo_path="./repo")
        self.web_retriever = DuckDuckGoSearchRun()
        
        # LLM Stack
        self.fast_llm = ChatOpenAI(model="gpt-3.5-turbo")
        self.powerful_llm = ChatOpenAI(model="gpt-4")
        
        # Tools
        self.tools = [
            SQLDatabaseTool(db_connection),
            SlackTool(token),
            JiraTool(api_key)
        ]
    
    def search(self, query: str, mode: str = "auto"):
        # 1. クエリ分類
        query_type = self._classify_query(query)
        
        # 2. 適切なRetriever選択
        if query_type == "code":
            docs = self.code_retriever.search(query)
        elif query_type == "document":
            docs = self.doc_retriever.search(query)
        else:
            docs = self.web_retriever.search(query)
        
        # 3. LLM選択（複雑さに応じて）
        llm = self.powerful_llm if query_type == "complex" else self.fast_llm
        
        # 4. 回答生成
        response = llm.invoke([
            SystemMessage(content="企業検索アシスタント"),
            HumanMessage(content=f"Context: {docs}\n\nQuestion: {query}")
        ])
        
        return response.content
    
    def _classify_query(self, query: str) -> str:
        # クエリ分類ロジック
        classification = self.fast_llm.invoke(
            f"Classify this query type: {query}\nTypes: code, document, web, complex"
        )
        return classification.content.strip()