LangChain with Custom Tool - Real-time stock price search - Part II (250821)

메모

우선주가 있는 경우, 정확한 결과를 반환하지 못하는 문제가 있음(250820)
get_korean_stock_price 호출 시 오류가 가끔씩 발생 -> 반복 실행(250820)
의문점 :
LLM은 어떻게 쿼리에서 적절한 단어를 선택하여 tool로 전달하게 되는가? (250820)

# Custom tool : 에이전트에 새로운 능력(@tool) 추가하기

from langchain.agents import initialize_agent, AgentType
# from langchain.llms import OpenAI
from langchain.agents.tools import tool

# 새로운 능력 #1 : @tool - 실시간 주가 검색하기
# from pydantic import BaseModel            # MCP 코드에서 수정
import yfinance as yf

# 함수 정의 및 데코레이션 적용
@tool
def get_korean_stock_price(symbol: str) -> str:
    """
    한국 주식의 종목 코드를 입력하면 현재 가격을 조회합니다.
    예시: '005930.KS'
    """
    ticker = symbol.strip()

    try:
        stock = yf.Ticker(ticker)
        price = stock.fast_info.get("last_price") or stock.fast_info.get("regular_market_price")
        name = stock.fast_info.get("shortName") or ticker

        if price is None:
            df = stock.history(period="1d", interval="1m")
            if not df.empty:
                price = df["Close"].iloc[-1]
            else:
                raise ValueError("가격 정보를 가져올 수 없습니다.")

        return f"{name}({ticker})의 현재 가격은 {price:.2f}원입니다."

    except Exception as e:
        return f"오류: {e}"

Google API 추가

• 구글 검색 API를 Tool로 추가하고 종목 코드를 직접 검색한다.

# 새로운 능력 #2 : 종목 코드를 Google에서 직접 검색하기
# 참고 : https://github.com/zarathucorp/blog/blob/master/source_code/Custom_Search_Zarathu.ipynb

import requests
import re
import pandas as pd
from langchain.agents.tools import tool  # LangChain Tool 데코레이터 사용

# 깃에 올릴 때 아래 삭제할 것
Google_SEARCH_ENGINE_ID = "60********"
Google_API_KEY = "AI*****************"

# 신뢰도 낮은 출처 제외
Trash_Link = [
    "tistory", "kin", "youtube", "blog", "book", "news", "dcinside",
    "fmkorea", "ruliweb", "theqoo", "clien", "mlbpark", "instiz", "todayhumor"
]

@tool
def Google_API(stock_name: str) -> str:
    """
    한국 주식 종목명을 구글에서 검색해 종목 코드(6자리)와 시장(KS/KQ)을 반환합니다.
    예) "삼성전자" -> "005930.KS"
        "셀트리온헬스케어" -> "091990.KQ"
    """
    wanted_row = 10  # 기본값 고정
    query = f"{stock_name} 종목코드 site:kr"
    query = query.replace("|", "OR") + " -filetype:pdf"

    df_google = pd.DataFrame(columns=['Title', 'Link', 'Description'])
    row_count = 0

    for start in range(1, 50, 10):  # 최대 50개 결과 탐색
        url = (
            f"https://www.googleapis.com/customsearch/v1"
            f"?key={Google_API_KEY}&cx={Google_SEARCH_ENGINE_ID}"
            f"&q={query}&start={start}"
        )
        data = requests.get(url).json()
        search_items = data.get("items", [])

        for i, item in enumerate(search_items, start=1):
            link = item.get("link", "")
            if any(trash in link for trash in Trash_Link):
                continue

            title = item.get("title", "")
            desc = item.get("snippet", "")
            df_google.loc[start + i] = [title, link, desc]
            row_count += 1

            # 종목 코드(숫자 6자리) 추출
            match = re.search(r"\b\d{6}\b", title + " " + desc + " " + link)
            if match:
                code = match.group(0)

                # 시장 구분 (코스피/코스닥 키워드 확인)
                text = (title + " " + desc + " " + link).lower()
                if "코스피" in text:
                    return f"{code}.KS"
                elif "코스닥" in text:
                    return f"{code}.KQ"
                else:
                    # 시장 구분 실패 시 기본값을 .KS로 반환
                    return f"{code}.KS"

            if row_count >= wanted_row:
                break

    # 결과 없을 때
    return "종목 코드를 찾을 수 없습니다."

# 에이전트 초기화 : 언어모델 / 도구 목록 / 에이전트 초기화 + 에이전트 실행

# 언어모델 
# from langchain_openai import ChatOpenAI
from langchain_community.chat_models import ChatOpenAI
from langchain_core.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
# pip install langchain-openai


llm = ChatOpenAI(
    base_url="http://localhost:1234/v1",
    api_key="lm-studio",
    # model="lmstudio-community/Meta-Llama-3.1-8B-Instruct-GGUF",
    model = "lmstudio-community/gemma-2-2b-it-GGUF",
    temperature=0.1,
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()], # 스트림 출력 콜백
)

# 도구 목록
tools = []
tools.append(Google_API)   # 메서드 이름
tools.append(get_korean_stock_price)   # 메서드 이름


# 에이전트 초기화
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=False
)

# 에이전트 실행
result = agent.run("삼성전자의 현재 주가를 조회해줘.")
print(result)

실행 결과

Thought: I need to get the stock code for Samsung Electronics first.
Action: Google_API
Action Input: 삼성전자Thought: Now that I have the stock code, I can use the get_korean_stock_price function to get the current price.
Action: get_korean_stock_price
Action Input: 005930.KSThought: I now know the final answer
Final Answer: 삼성전자의 현재 주가는 70550.00원입니다.
삼성전자의 현재 주가는 70550.00원입니다.