LangChain ile Çoklu Kaynaklı Arama Motoru Geliştirme

Yazar:komut.io ekibi
0 Yorum

Birden fazla kaynaktan bilgi toplayıp anlamlı bir cevap üretmek, modern yapay zeka uygulamalarının en çekici kullanım senaryolarından biri. Özellikle kurumsal ortamlarda “şirket içi dokümantasyon + harici API + web araması” gibi hibrit yapılar kurmanız gerektiğinde, LangChain’in çoklu araç mimarisi gerçekten işe yarıyor. Bu yazıda sıfırdan başlayıp üretim ortamına yakın bir çoklu kaynaklı arama motoru inşa edeceğiz.

Neden Çoklu Kaynak?

Tek bir LLM’e “bana bu konuyu anlat” dediğinizde, model eğitim verisinden ne biliyorsa onu söyler. Ama gerçek dünyada ihtiyaçlarınız çok daha spesifik: güncel fiyat bilgisi, şirket içi politika dokümanları, teknik knowledge base, belki de bir veritabanı sorgusu. Bunların hepsini aynı anda konuşturabilmek için LangChain’in Tool ve Agent yapısını kullanıyoruz.

Benim senaryom şuydu: Bir yazılım şirketinin iç destek botunu geliştiriyorduk. Kullanıcılar hem Confluence’daki dokümanlara, hem GitHub Issues’a, hem de genel web’e aynı anda soru sorabilmeliydi. Ayrı ayrı aramak yerine tek bir arayüzden hepsine ulaşmak hem kullanıcı deneyimini iyileştiriyor hem de destek ekibinin yükünü azaltıyor.

Ortam Kurulumu

Önce gerekli paketleri kuralım. Python 3.10+ öneriyorum, özellikle async yapılar için:

pip install langchain langchain-community langchain-openai
pip install faiss-cpu chromadb
pip install duckduckgo-search wikipedia
pip install python-dotenv requests beautifulsoup4
pip install tiktoken sentence-transformers

.env dosyanızı hazırlayın:

OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxx
SERPAPI_API_KEY=xxxxxxxxxxxxxxxxxx   # opsiyonel, DuckDuckGo ücretsiz alternatif
CONFLUENCE_URL=https://yourcompany.atlassian.net
CONFLUENCE_TOKEN=xxxxxxxxxxxxxxxxxx

Temel Agent Yapısı

LangChain’de çoklu kaynak araması için AgentExecutor kullanıyoruz. Agent, hangi aracı ne zaman kullanacağına kendisi karar veriyor. Bu kararı LLM veriyor, yani siz sadece araçları tanımlıyorsunuz:

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from dotenv import load_dotenv
import os

load_dotenv()

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY")
)

# System prompt - ajanın davranışını belirliyor
prompt = ChatPromptTemplate.from_messages([
    ("system", """Sen bir teknik destek asistanısın. Kullanıcının sorularını yanıtlamak için 
    elindeki araçları akıllıca kullan. Birden fazla kaynaktan bilgi topla ve 
    sonuçları sentezleyerek kapsamlı bir cevap ver. 
    Kaynak belirt: hangi bilgiyi nereden aldığını mutlaka söyle."""),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

Araç 1: Yerel Döküman Araması (RAG)

İlk araç, şirket içi dökümanlardan oluşan vector store’u sorgulayan bir RAG sistemi. Gerçek projede Confluence’dan çektiğimiz dokümanları FAISS’e yüklemiştik, burada yerel dosyalarla simüle ediyoruz:

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain.tools import Tool
from langchain.chains import RetrievalQA
from langchain_community.document_loaders import DirectoryLoader, TextFileLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

def create_document_search_tool():
    # Dökümanları yükle ve chunk'la
    loader = DirectoryLoader(
        "./docs",
        glob="**/*.txt",
        loader_cls=TextFileLoader
    )
    documents = loader.load()
    
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        separators=["nn", "n", " ", ""]
    )
    chunks = text_splitter.split_documents(documents)
    
    # Vector store oluştur
    embeddings = OpenAIEmbeddings()
    vectorstore = FAISS.from_documents(chunks, embeddings)
    vectorstore.save_local("./faiss_index")
    
    # Retrieval chain
    retriever = vectorstore.as_retriever(
        search_type="mmr",  # Maximum Marginal Relevance - çeşitlilik için
        search_kwargs={"k": 5, "fetch_k": 20}
    )
    
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True
    )
    
    def search_internal_docs(query: str) -> str:
        result = qa_chain.invoke({"query": query})
        sources = [doc.metadata.get("source", "bilinmiyor") 
                   for doc in result.get("source_documents", [])]
        return f"İç Doküman Sonucu: {result['result']}nKaynaklar: {', '.join(set(sources))}"
    
    return Tool(
        name="internal_document_search",
        description="""Şirket içi dokümanları, teknik kılavuzları ve politika belgelerini arar.
        Ürün özellikleri, teknik dokümantasyon veya şirket prosedürleri hakkında sorular için kullan.""",
        func=search_internal_docs
    )

search_type="mmr" kullanmak önemli bir detay. Basit similarity search yerine MMR, birbirine çok benzer chunk’ları tekrar etmek yerine çeşitli sonuçlar getiriyor. Özellikle uzun dokümanlarda bu fark büyük oluyor.

Araç 2: Web Araması

DuckDuckGo ücretsiz ve API key gerektirmiyor, prototip aşaması için ideal:

from langchain_community.tools import DuckDuckGoSearchRun, DuckDuckGoSearchResults
from langchain.tools import Tool

def create_web_search_tool():
    # DuckDuckGoSearchResults daha fazla metadata döndürüyor
    ddg_search = DuckDuckGoSearchResults(
        num_results=5,
        backend="text"  # "news" da kullanabilirsiniz
    )
    
    def web_search_with_context(query: str) -> str:
        try:
            results = ddg_search.run(query)
            return f"Web Arama Sonuçları:n{results}"
        except Exception as e:
            # Rate limit veya bağlantı hatası durumunda
            return f"Web araması şu an kullanılamıyor: {str(e)}"
    
    return Tool(
        name="web_search",
        description="""İnternette güncel bilgi arar. Güncel haberler, yeni teknolojiler,
        son gelişmeler veya iç dokümanlarda olmayan konular için kullan.
        Şirket içi bilgi yeterli olmadığında bu araca başvur.""",
        func=web_search_with_context
    )

Araç 3: Wikipedia Araması

Teknik terimler, genel kavramlar ve bağlam sağlamak için Wikipedia aracı işe yarıyor. Özellikle kullanıcının terminoloji konusunda kafasının karışık olduğu durumlarda:

from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
from langchain.tools import Tool

def create_wikipedia_tool():
    wikipedia = WikipediaAPIWrapper(
        lang="tr",  # Önce Türkçe dene
        top_k_results=2,
        doc_content_chars_max=2000
    )
    
    wiki_tool = WikipediaQueryRun(api_wrapper=wikipedia)
    
    def smart_wiki_search(query: str) -> str:
        # Önce Türkçe dene, yoksa İngilizce'ye geç
        try:
            result = wiki_tool.run(query)
            if "No good Wikipedia Search Result was found" in result or len(result) < 100:
                # İngilizce'ye geç
                wikipedia_en = WikipediaAPIWrapper(lang="en", top_k_results=2)
                wiki_en = WikipediaQueryRun(api_wrapper=wikipedia_en)
                result = wiki_en.run(query)
                result = f"[İngilizce Wikipedia]n{result}"
            return f"Wikipedia: {result}"
        except Exception as e:
            return f"Wikipedia araması başarısız: {str(e)}"
    
    return Tool(
        name="wikipedia_search",
        description="""Genel bilgi, tanımlar ve kavramsal açıklamalar için Wikipedia'yı arar.
        Teknik terimler, teknoloji geçmişi veya genel kavramlar için kullan.""",
        func=smart_wiki_search
    )

Araç 4: Özel API Entegrasyonu

Gerçek projelerde genellikle bir iç API’ye de bağlanmanız gerekiyor. Burada örnek olarak bir “sistem durumu” API’si simüle ediyorum:

import requests
import json
from langchain.tools import Tool
from datetime import datetime

def create_system_status_tool():
    def check_system_status(query: str) -> str:
        """
        Gerçek projede bu fonksiyon şirketin monitoring API'sine bağlanır.
        Burada mock data kullanıyoruz.
        """
        mock_status = {
            "timestamp": datetime.now().isoformat(),
            "services": {
                "api_gateway": {"status": "healthy", "latency_ms": 45},
                "database": {"status": "healthy", "connections": 127},
                "cache": {"status": "degraded", "hit_rate": 0.72},
                "message_queue": {"status": "healthy", "queue_depth": 234}
            },
            "recent_incidents": [
                {
                    "id": "INC-2847",
                    "title": "Cache servis yavaşlaması",
                    "status": "investigating",
                    "created_at": "2024-01-15T14:30:00Z"
                }
            ]
        }
        
        # Sorguya göre filtrele
        query_lower = query.lower()
        if "cache" in query_lower:
            cache_info = mock_status["services"]["cache"]
            return f"Cache durumu: {cache_info['status']}, Hit rate: {cache_info['hit_rate']:.0%}"
        
        if "incident" in query_lower or "sorun" in query_lower:
            incidents = mock_status["recent_incidents"]
            if incidents:
                inc = incidents[0]
                return f"Aktif Sorun: {inc['title']} (Durum: {inc['status']}, ID: {inc['id']})"
            return "Şu an aktif sorun bulunmuyor."
        
        # Genel durum özeti
        healthy = sum(1 for s in mock_status["services"].values() 
                     if s["status"] == "healthy")
        total = len(mock_status["services"])
        return (f"Sistem Durumu: {healthy}/{total} servis sağlıklı. "
                f"Son güncelleme: {mock_status['timestamp']}")
    
    return Tool(
        name="system_status",
        description="""Sistem servislerinin anlık durumunu, aktif olayları ve 
        performans metriklerini sorgular. Sistem sorunları, servis durumu veya 
        aktif incident'lar için kullan.""",
        func=check_system_status
    )

Her Şeyi Bir Araya Getirmek

Tüm araçları agent’a bağlıyoruz. create_openai_tools_agent OpenAI’nin function calling özelliğini kullanıyor, bu da ReAct agent’a kıyasla çok daha güvenilir araç seçimi sağlıyor:

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.messages import HumanMessage, AIMessage

# Araçları oluştur
tools = [
    create_document_search_tool(),
    create_web_search_tool(),
    create_wikipedia_tool(),
    create_system_status_tool()
]

# Agent oluştur
agent = create_openai_tools_agent(llm, tools, prompt)

# AgentExecutor - araç çağrılarını yönetiyor
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,          # Hangi araçların kullanıldığını görmek için
    max_iterations=5,      # Sonsuz döngüyü önle
    max_execution_time=30, # Saniye cinsinden timeout
    handle_parsing_errors=True,
    return_intermediate_steps=True  # Debug için
)

# Konuşma geçmişi ile kullan
def ask_with_history(question: str, chat_history: list = None):
    if chat_history is None:
        chat_history = []
    
    result = agent_executor.invoke({
        "input": question,
        "chat_history": chat_history
    })
    
    # Hangi araçlar kullanıldı?
    used_tools = []
    for step in result.get("intermediate_steps", []):
        tool_name = step[0].tool
        if tool_name not in used_tools:
            used_tools.append(tool_name)
    
    return {
        "answer": result["output"],
        "sources_used": used_tools,
        "steps": len(result.get("intermediate_steps", []))
    }

# Test
if __name__ == "__main__":
    history = []
    
    response = ask_with_history(
        "Cache servisi neden yavaş? Bu sorun ne zaman başladı ve nasıl çözebilirim?"
    )
    print(f"Cevap: {response['answer']}")
    print(f"Kullanılan araçlar: {response['sources_used']}")

Streaming ile Gerçek Zamanlı Yanıt

Kullanıcı arayüzü geliştiriyorsanız streaming şart. LangChain’de callback sistemi ile bunu kolayca yapabilirsiniz:

from langchain_core.callbacks import StreamingStdOutCallbackHandler
from langchain.callbacks.manager import CallbackManager

# Streaming LLM
streaming_llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    streaming=True,
    callback_manager=CallbackManager([StreamingStdOutCallbackHandler()])
)

# FastAPI ile kullanım örneği
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def generate_response(question: str):
    """Async generator - token token gönderir"""
    async for chunk in agent_executor.astream(
        {"input": question, "chat_history": []}
    ):
        if "output" in chunk:
            yield f"data: {chunk['output']}nn"
        elif "actions" in chunk:
            for action in chunk["actions"]:
                yield f"data: [Araç kullanılıyor: {action.tool}]nn"

@app.get("/search")
async def search_endpoint(q: str):
    return StreamingResponse(
        generate_response(q),
        media_type="text/event-stream"
    )

Performans ve Maliyet Optimizasyonu

Üretim ortamına geçerken dikkat etmeniz gereken birkaç konu var.

Caching: Aynı soruya her seferinde API çağrısı yapmak hem maliyetli hem de yavaş. LangChain’in semantic cache özelliğini kullanın:

from langchain.globals import set_llm_cache
from langchain_community.cache import InMemoryCache, SQLiteCache

# Development için in-memory cache
set_llm_cache(InMemoryCache())

# Production için SQLite (Redis de kullanabilirsiniz)
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

# Artık aynı veya çok benzer sorgular için LLM çağrısı yapılmaz
# FAISS ile semantic similarity kullanmak için:
from langchain_openai import OpenAIEmbeddings
from langchain_community.cache import GPTCache

# GPTCache ile semantic caching
# pip install gptcache

Araç Seçimini Kısıtlamak: Bazı sorular için gereksiz araç çağrısı yapılmasını önleyin. Bunun için tool description’larını iyi yazmak kritik. Belirsiz description’lar agent’ın yanlış araç seçmesine neden oluyor.

Token Yönetimi: Her araç yanıtını max_tokens ile sınırlayın:

from langchain.schema import Document
from langchain.text_splitter import TokenTextSplitter

def truncate_tool_output(output: str, max_tokens: int = 500) -> str:
    """Araç çıktısını token limitine göre kırp"""
    splitter = TokenTextSplitter(
        chunk_size=max_tokens,
        chunk_overlap=0,
        model_name="gpt-4o-mini"
    )
    chunks = splitter.split_text(output)
    return chunks[0] if chunks else output

Gerçek Dünya Dersleri

Birkaç ay bu sistemleri kurumsal ortamda işletirken öğrendiklerim:

  • Agent sonsuz döngüye girebilir. max_iterations=5 ve max_execution_time=30 gibi sınırları mutlaka koyun. Aksi halde bir araç hata döndürdüğünde agent aynı aracı defalarca deniyor.
  • Tool description’ı kritik. Agent neyi ne zaman kullanacağına description’a bakarak karar veriyor. “Genel bilgi için kullan” gibi muğlak açıklamalar yerine kesin sınırlar belirleyin. “Bu aracı SADECE şu durumlarda kullan” formatı daha iyi sonuç veriyor.
  • Verbose modu prod’da kapatın. verbose=True her adımı stdout’a yazıyor. Prod ortamda bunu loglama sistemine yönlendirin.
  • Araç başarısızlıklarını graceful handle edin. Her araç fonksiyonunuzda try/except olmalı ve anlamlı hata mesajı dönmeli. Agent hata mesajını da bilgi olarak işleyebiliyor.
  • Paralel araç çağrısı mümkün. OpenAI tools API, birden fazla aracı aynı anda çağırabilir. Bu latency’yi dramatik düşürüyor. create_openai_tools_agent bunu destekliyor.

Sonuç

LangChain ile çoklu kaynaklı arama motoru kurmak, aslında karmaşık görünen ama doğru araçlarla oldukça yönetilebilir bir iş. Önemli olan parçaları şöyle sıralayabilirim: her kaynağı iyi izole edilmiş bir Tool olarak modellemek, tool description’larını LLM’in anlayabileceği şekilde yazmak, üretim ortamı için caching ve rate limiting mekanizmalarını baştan planlamak.

Bu mimarinin güzel tarafı genişletilebilirliği. Yeni bir kaynak eklemek istediğinizde sadece yeni bir Tool yazıp listeye ekliyorsunuz, geri kalan her şey otomatik çalışıyor. Bir SQL veritabanı, bir Slack kanalı, bir Git repository hepsi aynı şekilde entegre olabiliyor.

Bir sonraki adım olarak LangGraph ile daha deterministik akışlar kurmayı veya multi-agent mimarilerde farklı uzman agent’ların koordinasyonunu inceleyebilirsiniz. Ama tek agent ile bile bu senaryoda gerçekten üretim kalitesi bir sistem kurabiliyorsunuz.

Bir yanıt yazın

E-posta adresiniz yayınlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir