LangChain with Custom Tool - Real-time stock price search (250819)

에이전트에 새로운 능력 추가하기

• initialize_agent
• AgentType

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

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

새로운 능력 : 실시간 주가 검색 기능(yfinance)

• LLM에서 종목 코드를 조회하는 기능이 없음 (향후 추가 필요)
• 코드 매핑 딕셔너리를 임의로 추가 : "삼성전자"를 검색하는 예제

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

# class StockRequest(BaseModel):            # MCP 코드에서 수정
#    symbol: str  # 종목 코드


# 종목명 → 종목코드 매핑 딕셔너리
# LLM에서 tool로 종목 코드가 전달되지 않는 문제 발생 (250819)
KOREAN_STOCKS = {
    "삼성전자": "005930.KS",   # 코스피 종목
    "삼성전자우": "005935.KS",
    "LG에너지솔루션": "373220.KS",
    "카카오": "035720.KQ",   # 코스닥 종목
    "네이버": "035420.KS",
}


# 함수 정의 및 데코레이션 적용
@tool
#def get_korean_stock_price(input: StockRequest) -> str:   # MCP 코드에서 수정
def get_korean_stock_price(input: str) -> str:
    """
    yfinance를 사용하여 한국 주식의 가격을 조회합니다.
    최대한 안정적이고 빠른 방법을 사용합니다.
    """
    # code = input.symbol.strip()
    code = input.strip()

    if code in KOREAN_STOCKS:
        ticker = KOREAN_STOCKS[code]
    else:
        # 2) 이미 접미사가 붙어있으면 그대로 사용
        if code.endswith(".KS") or code.endswith(".KQ"):
            ticker = code
        else:
            # 3) 우선 .KS (코스피) 조회 시도
            ticker = code + ".KS"
            test = yf.Ticker(ticker)
            if test.history(period="1d").empty:
                # 4) .KS에 데이터 없으면 .KQ로 시도
                ticker = code + ".KQ"

    try:
        stock = yf.Ticker(ticker)

        # 우선 fast_info를 먼저 사용
        price = stock.fast_info.get("last_price") or stock.fast_info.get("regular_market_price")
        name = stock.fast_info.get("shortName") or ticker

        # fast_info 실패 시 history 사용 (최근 1분봉 데이터)
        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}"

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

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

# 언어모델 
# 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(get_korean_stock_price)   # 메서드 이름

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

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

실행 결과

Entering new AgentExecutor chain...
Thought: I need to get the current price of Samsung Electronics.
Action: get_korean_stock_price
Action Input: 삼성전자Thought: I need to get the current price of Samsung Electronics.
Action: get_korean_stock_price
Action Input: 삼성전자
Observation: 005930.KS(005930.KS)의 현재 가격은 70050.00원입니다.
Thought:Thought: I now have the current price of Samsung Electronics.
Final Answer: 삼성전자의 현재 주가는 70050.00원입니다.
Thought: I now have the current price of Samsung Electronics.
Final Answer: 삼성전자의 현재 주가는 70050.00원입니다.
Finished chain.
삼성전자의 현재 주가는 70050.00원입니다.