RAG 프로젝트 - Google Search agent

🐹 검색엔진 선정 - Google Search agent 

네이버, 카카오 등과 같은 검색엔진을 비교하여 google search를 선정하였다.

다른 검색 엔진에 대한 내용은 아래에서 확인할 수 있다.

https://pleasestudy-alswldi.tistory.com/341

RAG 시스템 검색 엔진 정리

 

 

Google Search agent 사용 방법

 

pip install llama-index-tools-google

pip install google-api-python-client

🔑API key 발급

https://console.cloud.google.com/

Google 클라우드 플랫폼

 

https://programmablesearchengine.google.com/

Programmable Search Engine by Google

 

1. Google Cloud Console(https://console.cloud.google.com/) 에서 프로젝트를 생성(이미 있다면 선택)
2. Custom Search API를 활성화
3. API 키를 생성(GoogleSearchToolSpec에서 사용할 key)
4. Google Programmable Search Engine(https://programmablesearchengine.google.com/controlpanel/all) 에서 새 검색 엔진을 생성
5. 생성된 검색 엔진의 ID를 복사 (GoogleSearchToolSpec에서 사용할 id(engine))

 

1️⃣ Google Custom Search API를 직접 사용

• google_search 메서드에서 build함수를 사용하여 Custom Search API 서비스 객체 생성
• 생성된 서비스 객체를 사용하여 검색 수행
• 검색 결과의 URL에서 콘텐츠를 추출하여 Document 객체로 변환
  • pip install readability-lxml HTML 문서에서 메인 콘텐츠를 추출/불필요한 요소 제거
• Google 검색 API 호출을 직접 구현하므로 API의 매개변수 조정, 추가 기능 구현 등이 자유롭고 확장성이 높음
• workflow 전체 코드

'''
query에 대한 google 검색 -> context 가 생성되어 llm의 input으로 들어감
'''

class WEBWorkflow(Workflow):
    def __init__(self):
        super().__init__()
        self.embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-m3")
        self.llm = HuggingFaceLLM(
            model_name="rtzr/ko-gemma-2-9b-it",
            tokenizer_name="rtzr/ko-gemma-2-9b-it",
            device_map="auto",
            model_kwargs={"torch_dtype": torch.bfloat16},
            generate_kwargs={"max_length": 4096, "temperature": 0.7, "do_sample": True, "top_p": 0.95},
            max_new_tokens=1024,
        )
        self.google_api_key = "key"
        self.google_cse_id = "id"

    def fetch_clean_content(self, url):
        """URL에서 콘텐츠를 가져와 텍스트로 반환 (HTML 태그 제거)"""
        try:
            response = requests.get(url, timeout=10)
            response.raise_for_status()

            if not response.text.strip():
                print(f"Empty response for URL: {url}")
                return ""

            try:
                doc = ReadabilityDocument(response.text)
                summary_html = doc.summary()
                soup = BeautifulSoup(summary_html, 'html.parser')
                return soup.get_text(separator=" ").strip()
            except Exception as e:
                print(f"Readability failed for URL: {url}, falling back to BeautifulSoup. Error: {e}")
                soup = BeautifulSoup(response.text, 'html.parser')
                return soup.get_text(separator=" ").strip()
        except requests.exceptions.RequestException as e:
            print(f"Error fetching URL {url}: {e}")
            return ""

    def google_search(self, query: str) -> list[Document]:
        """Google Search API를 사용하여 결과를 가져오고 Document 리스트로 변환"""
        service = build("customsearch", "v1", developerKey=self.google_api_key) # 서비스 객체 생성
        res = service.cse().list(q=query, cx=self.google_cse_id, num=10).execute()

        documents = []
        for item in res.get('items', []):
            url = item.get('link')
            print(f"Fetching content from URL: {url}")
            content = self.fetch_clean_content(url)
            if content:
                documents.append(Document(text=content))
        return documents

    def load_documents(self) -> list[Document]:
        """Google Search를 통해 문서 리스트를 로드"""
        documents = self.google_search(self.query)  # Google Search 실행
        return documents

    @step()
    async def set_query(self, ctx: Context, ev: StartEvent) -> QueryEvent:
        """사용자 쿼리 설정"""
        query = "젠킨스 통신 에러"  # 사용자 입력
        await ctx.set("query", query)
        self.query = query  # 클래스 변수에 쿼리 저장
        print(f"Query set: {query}")
        return QueryEvent(query=query)

    @step()
    async def load_and_index(self, ctx: Context, ev: QueryEvent) -> IndexEvent:
        """문서 로드 및 인덱싱"""
        print("Loading documents from Google Search...")
        documents = self.load_documents()  # Google Search 데이터 로드
        
        index = VectorStoreIndex.from_documents(
            documents,
            embed_model=self.embed_model
        )
        print("Index created.")
        await ctx.set("index", index)
        return IndexEvent(index=index)

    @step()
    async def generate_prompt(self, ctx: Context, ev: IndexEvent) -> PromptEvent:
        """프롬프트 생성"""
        query = await ctx.get("query")
        prompt = (
            f"현재 문제는 {query}\n"
            " 구글에서 비슷한 문제를 찾아보고 이를 참고해서 어떻게 해야 할지 알려줘.\n\n"
        )
        print(f"Generated Prompt: {prompt}")
        return PromptEvent(prompt=prompt)

    @step()
    async def generate_response(self, ctx: Context, ev: PromptEvent) -> StopEvent:
        """응답 생성"""
        index = await ctx.get("index")
        query_engine = index.as_query_engine(llm=self.llm)
        response = query_engine.query(ev.prompt)
        print("Query response generated.")
        return StopEvent(result=str(response))

2️⃣ GoogleSearchToolSpec 를 이용

• GoogleSearchToolSpec은 llama_index 라이브러리에서 제공하는 Google 검색 도구
• GoogleSearchToolSpec은 검색 결과를 직접 Document 객체로 변환하여 제공
  • url과 같은 원본 정보를 확인하기 어려움
• GoogleSearchToolSpec을 사용하여 Google 검색 API를 간단히 통합하고, RAG 워크플로우 개발을 빠르게 진행할 수 있음
• BeautifulSoup 과 같이 html 코드를 parsing할 필요 없음
• workflow 전체 코드

class WEBWorkflow(Workflow):
    def __init__(self):
        super().__init__()
        self.embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-m3")
        self.llm = HuggingFaceLLM(
            model_name="rtzr/ko-gemma-2-9b-it",
            tokenizer_name="rtzr/ko-gemma-2-9b-it",
            device_map="auto",
            model_kwargs={"torch_dtype": torch.bfloat16},
            generate_kwargs={"temperature": 0.7, "do_sample": True, "top_p": 0.95},
            max_new_tokens=1024,
        )
        self.google_tool = GoogleSearchToolSpec(key="key", engine="id", num=10)

    def load_documents(self) -> list[Document]:
        """GoogleSearchToolSpec을 사용하여 검색 결과를 Document 리스트로 반환"""
        documents = self.google_tool.google_search(query=self.query)
        return documents

    @step()
    async def set_query(self, ctx: Context, ev: StartEvent) -> QueryEvent:
        """사용자 쿼리 설정"""
        query = "젠킨스 통신에러"
        await ctx.set("query", query)
        self.query = query
        return QueryEvent(query=query)

    @step()
    async def load_and_index(self, ctx: Context, ev: QueryEvent) -> IndexEvent:
        """문서 로드 및 인덱싱"""
        documents = self.load_documents()
        index = VectorStoreIndex.from_documents(documents, embed_model=self.embed_model)
        await ctx.set("index", index)
        return IndexEvent(index=index)

    @step()
    async def generate_prompt(self, ctx: Context, ev: IndexEvent) -> PromptEvent:
        """프롬프트 생성"""
        query = await ctx.get("query")
        prompt = (
            f"현재 문제는 {query}\n"
            " 구글에서 비슷한 문제를 찾아보고 이를 참고해서 어떻게 해야 할지 한국어로 알려줘.\n\n"
        )
        return PromptEvent(prompt=prompt)

    @step()
    async def generate_response(self, ctx: Context, ev: PromptEvent) -> StopEvent:
        """응답 생성"""
        index = await ctx.get("index")
        query_engine = index.as_query_engine(llm=self.llm)
        response = query_engine.query(ev.prompt)
        return StopEvent(result=str(response))

 

 

 

참고

https://github.com/run-llama/llama_index/blob/main/llama-index-integrations/tools/llama-index-tools-

google/examples/advanced_tools_usage.ipynb

llama_index/llama-index-integrations/tools/llama-index-tools-google/examples/advanced_tools_usage.ipynb at main · run-llama/lla

 

https://docs.llamaindex.ai/en/stable/api_reference/tools/google/#llama_index.tools.google.GoogleSearchToolSpec

Google - LlamaIndex

 

 

error

daum search test 진행 중 발생

Traceback (most recent call last):
  File "/home/work/cursor-data/llama-index-rag-system/workflows/web_search/workflow_web_search.py", line 41, in <module>
    content = fetch_clean_content(document['url'])
  File "/home/work/cursor-data/llama-index-rag-system/workflows/web_search/workflow_web_search.py", line 13, in fetch_clean_content
    return doc.summary()
  File "/home/work/.local/lib/python3.10/site-packages/readability/readability.py", line 268, in summary
    raise_with_traceback(Unparseable, sys.exc_info()[2], str_(e))
  File "/home/work/.local/lib/python3.10/site-packages/readability/compat/three.py", line 6, in raise_with_traceback
    raise exc_type(*args, **kwargs).with_traceback(traceback)
  File "/home/work/.local/lib/python3.10/site-packages/readability/readability.py", line 213, in summary
    self._html(True)
  File "/home/work/.local/lib/python3.10/site-packages/readability/readability.py", line 148, in _html
    self.html = self._parse(self.input)
  File "/home/work/.local/lib/python3.10/site-packages/readability/readability.py", line 157, in _parse
    doc, self.encoding = build_doc(input)
  File "/home/work/.local/lib/python3.10/site-packages/readability/htmls.py", line 21, in build_doc
    doc = lxml.html.document_fromstring(
  File "/usr/local/lib/python3.10/dist-packages/lxml/html/__init__.py", line 761, in document_fromstring
    raise etree.ParserError(
readability.readability.Unparseable: Document is empty

readability-lxml 라이브러리가 처리하려는 문서가 비어 있거나(HTML 콘텐츠가 없음) 구조적으로 파싱할 수 없는 경우 발생

• 빈 응답 본문 처리:
  • HTTP 응답 본문이 비어 있을 경우 예외 처리를 추가.
• 요약 실패 시 대체 처리:
  • readability.Document에서 예외가 발생하면 BeautifulSoup을 사용하여 대체로 텍스트를 추출.