조선왕조실록 검색 agent

from operator import itemgetter

from langchain_openai import ChatOpenAI
from langchain.agents import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents.format_scratchpad.openai_tools import format_to_openai_tool_messages
from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParser
from langchain.agents import AgentExecutor
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
import requests
from bs4 import BeautifulSoup
import re
from urllib3.exceptions import InsecureRequestWarning
from typing import List, Tuple

# type : k (국역)
input_question = "기자조선에 대해서 검색 결과 5개를 찾아 줘. type은 k입니다."

llm = ChatOpenAI()

# 조선왕조실록 검색 크롤링 tool
@tool
def search_sillok(keyword: str, pageIndex: int = 1, pageUnit: int = 100, type: str = "k") -> List[Tuple[str, str]]:
    """
    Search the Sillok database and return a list of URLs and titles.

    :param keyword: Search keyword
    :param pageIndex: Page number to fetch (default: 1)
    :param pageUnit: Number of results per page (default: 100)
    :param type: Search type, 'k' for translated, 'w' for original text (default: 'k')
    :return: List of tuples containing (URL, title) for each result
    """
    # SSL warning disable
    requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

    # URL setup
    url = "https://sillok.history.go.kr/search/searchResultList.do"

    # Headers setup
    headers = {
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
        "Accept-Encoding": "gzip, deflate, br, zstd",
        "Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7",
        "Cache-Control": "max-age=0",
        "Connection": "keep-alive",
        "Content-Type": "application/x-www-form-urlencoded",
        "Cookie": f"activexime_enable=false; JSESSIONID=9DC5C42802E9959B05B9057A55843C6C; PCID=17199313473597572209955; RC_RESOLUTION=1920*1080; RC_COLOR=24; sillok_op_search_pageUnit={pageUnit}; sillok_view_scroll=100",
        "Host": "sillok.history.go.kr",
        "Origin": "https://sillok.history.go.kr",
        "Sec-Ch-Ua": '"Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"',
        "Sec-Ch-Ua-Mobile": "?0",
        "Sec-Ch-Ua-Platform": '"Windows"',
        "Sec-Fetch-Dest": "document",
        "Sec-Fetch-Mode": "navigate",
        "Sec-Fetch-Site": "same-origin",
        "Sec-Fetch-User": "?1",
        "Upgrade-Insecure-Requests": "1",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36",
    }

    # POST request payload
    payload = {
        "topSearchWord": keyword,
        "pageIndex": pageIndex,
        "initPageUnit": "0",
        "type": type,
        "sillokType": "S",
        "topSearchWord_ime": f'<span class="newbatang">{keyword}</span>',
    }

    def clean_text(text: str) -> str:
        cleaned = re.sub(r"\s+", " ", text)
        return cleaned.strip()

    def create_full_url(id_info: str) -> str:
        return f"https://sillok.history.go.kr/id/{id_info}"

    # Send POST request
    response = requests.post(url, headers=headers, data=payload, verify=False)

    if response.status_code != 200:
        raise Exception(f"Request failed. Status code: {response.status_code}")

    # Parse HTML
    soup = BeautifulSoup(response.text, "html.parser")

    # Find all <a> tags
    links = soup.find_all("a", href=re.compile("javascript:goView"))

    # Extract information from links
    results = []
    for link in links:
        match = re.search(r"goView\('([^']+)',", link["href"])
        if match:
            id_info = match.group(1)
            full_url = create_full_url(id_info)
            title = clean_text(link.text)
            results.append((full_url, title))

    return results

tools= [search_sillok]

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are very powerful assistant, but don't know current events",
        ),
        ("user", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ]
)

llm_with_tools = llm.bind_tools(tools)


input_generator = RunnableParallel(
    {
        "input": itemgetter("input"),
        "agent_scratchpad": lambda x: format_to_openai_tool_messages(
            x["intermediate_steps"]
        ),
    }
)

agent = (
    input_generator
    | prompt
    | llm_with_tools
    | OpenAIToolsAgentOutputParser()
)

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=False)

result = agent_executor.invoke({"input": input_question})

print('------------------------------------------')
print('Question : ', input_question)
print('Answer   : ', result['output'])
print('------------------------------------------')