LLM 개발

# pip install anthropic python-dotenv

import anthropic, os
from dotenv import load_dotenv

load_dotenv()
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Python으로 수급 분석 시스템을 만들려면?"}
    ],
)
print(message.content[0].text)

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "CQRS 패턴을 설명해주세요."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    print()

final = stream.get_final_message()
print(f"\n토큰: {final.usage.input_tokens} in / {final.usage.output_tokens} out")

import json

tools = [{
    "name": "get_stock_price",
    "description": "종목의 현재가와 외인 수급 정보를 가져옵니다.",
    "input_schema": {
        "type": "object",
        "properties": {"code": {"type": "string", "description": "종목코드"}},
        "required": ["code"],
    },
}]

def get_stock_price(code: str) -> dict:
    return {"code": code, "price": 78000, "foreign_net": 1500000}

messages = [{"role": "user", "content": "삼성전자(005930) 지금 사도 될까요?"}]

while True:
    resp = client.messages.create(
        model="claude-sonnet-4-6", max_tokens=1024,
        tools=tools, messages=messages,
    )
    messages.append({"role": "assistant", "content": resp.content})
    if resp.stop_reason == "end_turn":
        print(next(b.text for b in resp.content if hasattr(b, "text")))
        break
    tool_results = [
        {"type": "tool_result", "tool_use_id": b.id,
         "content": json.dumps(get_stock_price(**b.input), ensure_ascii=False)}
        for b in resp.content if b.type == "tool_use"
    ]
    messages.append({"role": "user", "content": tool_results})

# pip install langchain langchain-anthropic

from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatAnthropic(model="claude-sonnet-4-6", max_tokens=1024)

prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 {domain} 전문가입니다. 핵심만 간결하게 답변하세요."),
    ("human", "{question}"),
])

chain = prompt | llm | StrOutputParser()

result = chain.invoke({"domain": "백엔드 아키텍처", "question": "CQRS와 이벤트 소싱의 차이는?"})
print(result)

# pip install langchain-community chromadb sentence-transformers

from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

# 문서 인덱싱
chunks = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50).split_documents(
    TextLoader("knowledge_base.txt", encoding="utf-8").load()
)
db = Chroma.from_documents(chunks, HuggingFaceEmbeddings(model_name="BAAI/bge-m3"), persist_directory="./db")
retriever = db.as_retriever(search_kwargs={"k": 4})

# RAG 체인
rag_prompt = ChatPromptTemplate.from_template("""
컨텍스트를 기반으로 답하세요. 없으면 "문서에 없습니다"라고 하세요.

컨텍스트:
{context}

질문: {question}
""")

rag_chain = (
    {"context": retriever | (lambda docs: "\n\n".join(d.page_content for d in docs)),
     "question": RunnablePassthrough()}
    | rag_prompt | llm | StrOutputParser()
)

print(rag_chain.invoke("t1702 API는 어떻게 사용하나요?"))