LLM 호출부터 체인까지

 1# check_env.py - 환경 점검 스크립트
 2import sys
 3import importlib
 4
 5print(f"Python 버전: {sys.version}")
 6assert sys.version_info >= (3, 10), "Python 3.10 이상이 필요합니다."
 7
 8packages = ["langchain", "langchain_core", "langchain_aws", "dotenv"]
 9for name in packages:
10    try:
11        mod = importlib.import_module(name)
12        version = getattr(mod, "__version__", "버전 정보 없음")
13        print(f"[OK] {name} ({version})")
14    except ImportError:
15        print(f"[실패] {name} 이 설치되지 않았습니다. pip install 을 확인하세요.")

 1# check_credentials.py - 자격 증명 로드 확인
 2import os
 3from dotenv import load_dotenv
 4
 5load_dotenv()  # .env 파일의 내용을 환경변수로 읽어 들입니다
 6
 7key = os.getenv("AWS_ACCESS_KEY_ID")
 8if key:
 9    # 키 전체를 출력하면 안 되므로 앞 4자리만 확인합니다
10    print(f"[OK] AWS_ACCESS_KEY_ID 로드됨: {key[:4]}...")
11else:
12    print("[실패] AWS_ACCESS_KEY_ID 가 없습니다. .env 파일 위치와 내용을 확인하세요.")

 1# first_call_raw.py - 원본 요청/응답 JSON 구조 확인
 2import json
 3import boto3
 4from dotenv import load_dotenv
 5
 6load_dotenv()
 7client = boto3.client("bedrock-runtime", region_name="us-east-1")
 8
 9# 요청: modelId + messages 배열이 핵심입니다
10response = client.converse(
11    modelId="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
12    messages=[
13        {"role": "user", "content": [{"text": "안녕하세요. 자기소개를 한 문장으로 해주세요."}]}
14    ],
15)
16
17# 응답 전체를 JSON으로 출력해 구조를 눈으로 확인합니다
18print(json.dumps(response, indent=2, ensure_ascii=False, default=str))

 1# messages_role.py - role별 동작 차이 확인
 2from dotenv import load_dotenv
 3from langchain_aws import ChatBedrockConverse
 4from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
 5
 6load_dotenv()
 7llm = ChatBedrockConverse(
 8    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
 9    region_name="us-east-1",
10)
11
12# 1) system 메시지 없이
13r1 = llm.invoke([HumanMessage(content="사내 메신저가 접속이 안 됩니다.")])
14print("[system 없음]", r1.content)
15
16# 2) system 메시지로 페르소나 지정
17r2 = llm.invoke([
18    SystemMessage(content="너는 삼성SDS IT지원팀 상담원이야. 접수 확인과 예상 처리 절차를 정중한 존댓말 두 문장으로 안내해."),
19    HumanMessage(content="사내 메신저가 접속이 안 됩니다."),
20])
21print("[system 있음]", r2.content)
22
23# 3) 대화 기록(assistant 메시지 포함)으로 맥락 전달
24r3 = llm.invoke([
25    SystemMessage(content="너는 삼성SDS IT지원팀 상담원이야."),
26    HumanMessage(content="사내 메신저가 접속이 안 됩니다."),
27    AIMessage(content="접수되었습니다. 사용 중인 운영체제를 알려주시겠어요?"),
28    HumanMessage(content="Windows 11이요."),  # 모델이 '메신저 장애' 맥락을 알고 답합니다
29])
30print("[대화 기록]", r3.content)

 1# tokens.py - 응답의 토큰 사용량 확인
 2from dotenv import load_dotenv
 3from langchain_aws import ChatBedrockConverse
 4
 5load_dotenv()
 6llm = ChatBedrockConverse(
 7    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
 8    region_name="us-east-1",
 9)
10
11result = llm.invoke("LangChain을 한 문장으로 설명해줘.")
12
13print("응답:", result.content)
14
15# 사용량은 호출 방식에 따라 안 실려 올 수 있으므로 먼저 있는지 확인합니다
16if result.usage_metadata:
17    usage = result.usage_metadata
18    print(f"입력 토큰: {usage['input_tokens']}")
19    print(f"출력 토큰: {usage['output_tokens']}")
20    print(f"전체 토큰: {usage['total_tokens']}")
21else:
22    print("이번 응답에는 토큰 사용량이 실려 오지 않았습니다.")

 1# params.py - temperature와 max_tokens 비교
 2from dotenv import load_dotenv
 3from langchain_aws import ChatBedrockConverse
 4
 5load_dotenv()
 6MODEL = "us.anthropic.claude-sonnet-4-5-20250929-v1:0"
 7
 8prompt = "삼성SDS 사내 카페 이름 아이디어를 하나만 제안해줘."
 9
10# temperature 0: 결정적인 출력. 반복 호출해도 거의 같은 답
11llm_t0 = ChatBedrockConverse(model=MODEL, region_name="us-east-1", temperature=0)
12# temperature 1: 다양성 높은 출력. 호출마다 다른 답
13llm_t1 = ChatBedrockConverse(model=MODEL, region_name="us-east-1", temperature=1)
14
15print("--- temperature=0 (3회 호출) ---")
16for _ in range(3):
17    print(llm_t0.invoke(prompt).content)
18
19print("--- temperature=1 (3회 호출) ---")
20for _ in range(3):
21    print(llm_t1.invoke(prompt).content)
22
23# max_tokens: 출력 길이 제한. 너무 작으면 답이 잘립니다
24llm_short = ChatBedrockConverse(model=MODEL, region_name="us-east-1", max_tokens=20)
25result = llm_short.invoke("LangChain의 장점을 설명해줘.")
26print("--- max_tokens=20 ---")
27print(result.content)
28print("종료 사유:", result.response_metadata.get("stopReason"))

 1# limits.py - LLM 한계 재현
 2from dotenv import load_dotenv
 3from langchain_aws import ChatBedrockConverse
 4
 5load_dotenv()
 6llm = ChatBedrockConverse(
 7    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
 8    region_name="us-east-1",
 9    temperature=0,
10)
11
12# 1) 환각: 존재하지 않는 사내 규정을 물으면 지어낼 수 있습니다
13print(llm.invoke("삼성SDS 연차 이월 규정 제12조 3항 내용을 알려줘.").content)
14
15# 2) 계산 취약: 큰 수 곱셈
16print(llm.invoke("1234567 곱하기 8901234는? 숫자만 답해.").content)
17print("실제 답:", 1234567 * 8901234)
18
19# 3) 상태 망각: 두 번의 독립 호출은 서로를 모릅니다
20llm.invoke("내 이름은 김하늘이야.")
21print(llm.invoke("내 이름이 뭐였지?").content)

 1# why_langchain.py - 직접 호출 vs LangChain 비교
 2import boto3
 3from dotenv import load_dotenv
 4from langchain_aws import ChatBedrockConverse
 5from langchain_core.messages import SystemMessage, HumanMessage
 6
 7load_dotenv()
 8MODEL = "us.anthropic.claude-sonnet-4-5-20250929-v1:0"
 9text = "LangChain is the future of AI development"
10
11# --- A. SDK 직접 호출: provider 전용 형식 ---
12client = boto3.client("bedrock-runtime", region_name="us-east-1")
13resp = client.converse(
14    modelId=MODEL,
15    system=[{"text": "너는 짧고 간결한 번역가야."}],
16    messages=[{"role": "user", "content": [{"text": f"{text}를 한국어로 번역해."}]}],
17)
18print("[SDK 직접]", resp["output"]["message"]["content"][0]["text"])
19
20# --- B. LangChain: 표준 인터페이스 ---
21messages = [
22    SystemMessage(content="너는 짧고 간결한 번역가야."),
23    HumanMessage(content=f"{text}를 한국어로 번역해."),
24]
25llm = ChatBedrockConverse(model=MODEL, region_name="us-east-1")
26print("[LangChain]", llm.invoke(messages).content)
27
28# 모델을 다른 provider로 바꿔도 messages와 invoke 코드는 그대로입니다
29# from langchain_openai import ChatOpenAI
30# llm = ChatOpenAI(model="gpt-5.4-mini")  # 동일하게 동작

 1# components.py - 세 부품을 각각 실행해 입출력 타입 확인
 2from dotenv import load_dotenv
 3from langchain_aws import ChatBedrockConverse
 4from langchain_core.prompts import PromptTemplate, ChatPromptTemplate
 5from langchain_core.output_parsers import StrOutputParser
 6
 7load_dotenv()
 8
 9# --- 부품 1: PromptTemplate ---
10# 입력: dict, 출력: 완성된 프롬프트 (PromptValue)
11prompt = PromptTemplate.from_template("{text}를 {language}로 번역해줘.")
12filled = prompt.invoke({"text": "Good morning", "language": "한국어"})
13print("1) prompt 출력 타입:", type(filled).__name__)
14print("   내용:", filled.to_string())
15
16# --- 부품 2: Model ---
17# 입력: PromptValue(또는 메시지, 문자열), 출력: AIMessage
18llm = ChatBedrockConverse(
19    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
20    region_name="us-east-1",
21)
22ai_message = llm.invoke(filled)
23print("2) llm 출력 타입:", type(ai_message).__name__)
24print("   내용:", ai_message.content)
25
26# --- 부품 3: OutputParser ---
27# 입력: AIMessage, 출력: str
28parser = StrOutputParser()
29text = parser.invoke(ai_message)
30print("3) parser 출력 타입:", type(text).__name__)
31print("   내용:", text)
32
33# --- 대화형 템플릿: ChatPromptTemplate ---
34chat_prompt = ChatPromptTemplate.from_messages([
35    ("system", "너는 삼성SDS {team} 담당자 톤으로 대답하는 어시스턴트야."),
36    ("human", "{question}"),
37])
38msgs = chat_prompt.invoke({"team": "IT지원팀", "question": "VPN 설정 방법 알려줘."})
39print("4) chat_prompt 출력:", msgs.to_messages())

 1# lcel_basic.py - 체인 구성과 invoke, stream, batch
 2from dotenv import load_dotenv
 3from langchain_aws import ChatBedrockConverse
 4from langchain_core.prompts import PromptTemplate
 5from langchain_core.output_parsers import StrOutputParser
 6
 7load_dotenv()
 8
 9prompt = PromptTemplate.from_template(
10    "너는 삼성SDS 사내 커뮤니케이션 담당자야. {topic}에 대한 공지 문구를 두 문장으로 써줘."
11)
12llm = ChatBedrockConverse(
13    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
14    region_name="us-east-1",
15)
16parser = StrOutputParser()
17
18# 파이프 하나로 조립: dict -> 프롬프트 -> AIMessage -> str
19chain = prompt | llm | parser
20
21# 1) invoke: 단건 호출
22print("[invoke]")
23print(chain.invoke({"topic": "사내 시스템 정기 점검"}))
24print("---")
25
26# 2) stream: 토큰 단위 실시간 출력 (챗봇 UI의 타이핑 효과)
27print("[stream] ", end="", flush=True)
28for chunk in chain.stream({"topic": "보안 교육 일정"}):
29    print(chunk, end="", flush=True)
30print("\n---")
31
32# 3) batch: 여러 입력을 한꺼번에 병렬 처리
33inputs = [
34    {"topic": "주차장 공사"},
35    {"topic": "구내식당 메뉴 개편"},
36    {"topic": "사원증 재발급 안내"},
37]
38results = chain.batch(inputs)
39print("[batch]")
40for i, r in zip(inputs, results):
41    print(f"- {i['topic']}: {r[:30]}...")

 1# lcel_debug.py - 중간 단계 입출력 추적
 2from dotenv import load_dotenv
 3from langchain_aws import ChatBedrockConverse
 4from langchain_core.prompts import PromptTemplate
 5from langchain_core.output_parsers import StrOutputParser
 6from langchain_core.runnables import RunnableLambda
 7
 8load_dotenv()
 9
10def tap(label):
11    """값을 출력하고 그대로 다음 단계로 통과시키는 관찰 지점을 만듭니다."""
12    def _tap(value):
13        print(f"\n=== [{label}] 타입: {type(value).__name__} ===")
14        print(value)
15        return value  # 반드시 그대로 반환해야 체인이 이어집니다
16    return RunnableLambda(_tap)
17
18prompt = PromptTemplate.from_template("{word}의 반대말 하나만 답해.")
19llm = ChatBedrockConverse(
20    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
21    region_name="us-east-1",
22    temperature=0,
23)
24
25# 각 단계 사이에 관찰 지점을 끼워 넣습니다.
26chain = (
27    tap("입력")
28    | prompt
29    | tap("프롬프트 완성 후")
30    | llm
31    | tap("모델 응답 후")
32    | StrOutputParser()
33)
34
35result = chain.invoke({"word": "출근"})
36print("\n최종 결과:", result)

 1# structured_output.py - 응답을 JSON 스키마로 받기
 2from dotenv import load_dotenv
 3from pydantic import BaseModel, Field
 4from langchain_aws import ChatBedrockConverse
 5from langchain_core.prompts import PromptTemplate
 6from langchain_core.output_parsers import JsonOutputParser
 7
 8load_dotenv()
 9
10# 1) 출력 스키마 정의 - Field의 description이 모델에게 전달되는 설명입니다
11class InquiryAnalysis(BaseModel):
12    category: str = Field(description="문의 분류: 'IT지원', '인사', '총무' 중 하나")
13    urgency: str = Field(description="긴급도: '높음', '보통', '낮음' 중 하나")
14    summary: str = Field(description="문의 내용 한 줄 요약")
15
16# 2) 파서 생성 - 스키마 기반 형식 지침을 자동 생성합니다
17parser = JsonOutputParser(pydantic_object=InquiryAnalysis)
18
19# 3) 프롬프트에 형식 지침 끼워 넣기
20#    partial_variables: 사용자 입력이 아닌 값을 미리 채워 두는 자리입니다
21prompt = PromptTemplate(
22    template="다음 사내 문의를 분석해줘.\n\n문의: {inquiry}\n\n{format_instructions}",
23    input_variables=["inquiry"],
24    partial_variables={"format_instructions": parser.get_format_instructions()},
25)
26
27llm = ChatBedrockConverse(
28    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
29    region_name="us-east-1",
30    temperature=0,
31)
32chain = prompt | llm | parser
33
34result = chain.invoke({"inquiry": "노트북이 갑자기 안 켜지는데 오후에 고객 발표가 있습니다. 빨리 도와주세요."})
35
36print("타입:", type(result).__name__)
37print("전체:", result)
38print("분류만:", result["category"])

 1# inquiry_chain.py - 사내 문의 응답
 2from dotenv import load_dotenv
 3from pydantic import BaseModel, Field
 4from langchain_aws import ChatBedrockConverse
 5from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
 6from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
 7from langchain_core.runnables import RunnableParallel
 8
 9load_dotenv()
10llm = ChatBedrockConverse(
11    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
12    region_name="us-east-1",
13    temperature=0,
14)
15
16# --- 부품 A: 문의 분류 체인 (구조화 출력) ---
17class Routing(BaseModel):
18    department: str = Field(description="담당 부서: 'IT지원팀', '인사팀', '총무팀' 중 하나")
19    urgency: str = Field(description="긴급도: '높음', '보통', '낮음' 중 하나")
20
21routing_parser = JsonOutputParser(pydantic_object=Routing)
22routing_prompt = PromptTemplate(
23    template="다음 사내 문의의 담당 부서와 긴급도를 판단해.\n\n문의: {inquiry}\n\n{format_instructions}",
24    input_variables=["inquiry"],
25    partial_variables={"format_instructions": routing_parser.get_format_instructions()},
26)
27routing_chain = routing_prompt | llm | routing_parser
28
29# --- 부품 B: 답변 초안 체인 (문자열 출력) ---
30answer_prompt = ChatPromptTemplate.from_messages([
31    ("system",
32     "너는 삼성SDS 사내 헬프데스크 상담원이야. 정중한 존댓말로 답하고 "
33     "마지막 문장에 예상 처리 기한을 안내해. 세 문장 이내로 써."),
34    ("human", "{inquiry}"),
35])
36answer_chain = answer_prompt | llm | StrOutputParser()
37
38# --- 조립: 두 체인을 병렬 실행 후 하나의 dict로 합칩니다 ---
39full_chain = RunnableParallel(routing=routing_chain, draft=answer_chain)
40
41# --- 실행 ---
42# inquiry = {"inquiry": "재택근무 신청 절차가 어떻게 되나요? 다음 주부터 필요합니다."}
43inquiry = {"inquiry": "내일 당장 최사하고 싶습니다. 절차 알려주세요."}
44result = full_chain.invoke(inquiry)
45
46print("담당 부서:", result["routing"]["department"])
47print("긴급도:", result["routing"]["urgency"])
48print("답변 초안:")
49print(result["draft"])

 1# my_inquiry_chain.py - DevOps 툴 장애/요청 접수 체인 (Jenkins, GitHub, Nexus, ArgoCD, Harbor)
 2from dotenv import load_dotenv
 3from pydantic import BaseModel, Field
 4from langchain_aws import ChatBedrockConverse
 5from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
 6from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
 7from langchain_core.runnables import RunnableParallel
 8
 9load_dotenv()
10llm = ChatBedrockConverse(
11    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
12    region_name="us-east-1",
13    temperature=0,
14)
15
16# --- 부품 A: 접수 분류 체인 (구조화 출력) ---
17class DevOpsTicket(BaseModel):
18    severity: str = Field(
19        description="장애 등급: 'P1'(전사/다수 조직 서비스 중단), "
20                    "'P2'(일부 조직·기능 장애, 우회 가능), "
21                    "'P3'(단일 사용자 문제, 단순 요청·문의) 중 하나"
22    )
23    system: str = Field(
24        description="대상 시스템: 'Jenkins', 'GitHub', 'Nexus', 'ArgoCD', 'Harbor', '기타' 중 하나"
25    )
26    request_type: str = Field(
27        description="처리 유형: '장애', '권한요청', '설정변경', '사용문의' 중 하나"
28    )
29    summary: str = Field(description="접수 내용 한 줄 요약")
30
31ticket_parser = JsonOutputParser(pydantic_object=DevOpsTicket)
32ticket_prompt = PromptTemplate(
33    template=(
34        "다음은 삼성전자 DX 사업부 DevOps 툴 운영팀에 들어온 접수 건이야. "
35        "장애 등급, 대상 시스템, 처리 유형을 판단하고 한 줄로 요약해. "
36        "대상 시스템이 불분명하면 '기타'로 분류해.\n\n"
37        "접수 내용: {inquiry}\n\n{format_instructions}"
38    ),
39    input_variables=["inquiry"],
40    partial_variables={"format_instructions": ticket_parser.get_format_instructions()},
41)
42ticket_chain = ticket_prompt | llm | ticket_parser
43
44# --- 부품 B: 접수 회신 초안 체인 (문자열 출력) ---
45answer_prompt = ChatPromptTemplate.from_messages([
46    ("system",
47     "너는 삼성전자 DX 사업부 DevOps 툴 운영팀의 접수 담당자야. "
48     "Jenkins, GitHub, Nexus, ArgoCD, Harbor를 운영하고 있어. "
49     "정중하지만 간결한 업무 톤의 존댓말로 정확히 세 문장으로 답해.\n"
50     "1문장: 접수 확인 (대상 시스템과 현상을 짚어서)\n"
51     "2문장: 예상 대응 방향과 1차 회신 시점 (장애는 30분 이내, 요청·문의는 1영업일 이내)\n"
52     "3문장: 처리에 필요한 추가 정보 요청 (예: Job/Repository/Application 이름, 에러 로그, 발생 시각, 영향 범위)\n"
53     "추측으로 원인을 단정하지 말고, 목록 기호나 인사말 없이 문장만 써."),
54    ("human", "{inquiry}"),
55])
56answer_chain = answer_prompt | llm | StrOutputParser()
57
58# --- 조립: 두 체인을 병렬 실행 후 하나의 dict로 합칩니다 ---
59full_chain = RunnableParallel(ticket=ticket_chain, draft=answer_chain)
60
61# --- 실행: 서로 다른 입력 + 일부러 애매한 입력 ---
62inquiries = [
63    # 명확한 P1 장애
64    "지금 Jenkins 마스터 접속이 안 됩니다. 사업부 전체 빌드가 30분째 멈춰 있어요.",
65    # 권한 요청 (P3 예상)
66    "신규 입사자인데 GitHub의 dx-mobile 조직 repository 쓰기 권한 요청드립니다.",
67    # 일부 장애 (P2 예상)
68    "ArgoCD에서 stg 클러스터 앱 몇 개가 OutOfSync 상태로 계속 남아 있습니다. prod는 정상입니다.",
69    # 설정 변경
70    "Nexus에 사내 npm 프록시 저장소 하나 추가해 주실 수 있나요? 외부 registry 미러용입니다.",
71    # 애매한 입력: 시스템이 둘 이상 얽혀 있고 장애인지 설정 문제인지 불분명
72    "배포가 안 돼요. 이미지 pull 하다가 실패하는 것 같은데 어디 문제인지 모르겠어요.",
73    # 애매한 입력: 시스템도 현상도 불명확
74    "어제부터 뭔가 느려요. 급한 건 아닌데 확인 부탁드립니다.",
75]
76
77for inquiry in inquiries:
78    result = full_chain.invoke({"inquiry": inquiry})
79    ticket = result["ticket"]
80    print("=" * 70)
81    print("접수 내용:", inquiry)
82    print("장애 등급:", ticket["severity"])
83    print("대상 시스템:", ticket["system"])
84    print("처리 유형:", ticket["request_type"])
85    print("요약:", ticket["summary"])
86    print("회신 초안:")
87    print(result["draft"])

 1# multiturn_manual.py - chat_history 리스트 직접 관리로 원리 확인
 2from dotenv import load_dotenv
 3from langchain_aws import ChatBedrockConverse
 4from langchain_core.messages import HumanMessage
 5
 6load_dotenv()
 7llm = ChatBedrockConverse(
 8    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
 9    region_name="us-east-1",
10)
11
12chat_history = []  # 대화 기록을 우리가 직접 들고 있습니다
13
14for user_input in ["안녕, 내 이름은 김하늘이야.", "내 이름이 뭐였지?"]:
15    chat_history.append(HumanMessage(content=user_input))
16    response = llm.invoke(chat_history)   # 기록 전체를 매번 보냅니다
17    chat_history.append(response)         # AI 응답도 기록에 추가합니다
18    print(f"사용자: {user_input}")
19    print(f"AI: {response.content}")
20    print(f"(현재 기록 {len(chat_history)}개, 입력 토큰 {response.usage_metadata['input_tokens']})\n")

 1# multiturn_history.py - 권장 방식: RunnableWithMessageHistory
 2import warnings
 3
 4from dotenv import load_dotenv
 5from langchain_aws import ChatBedrockConverse
 6from langchain_core._api import LangChainDeprecationWarning
 7from langchain_core.chat_history import InMemoryChatMessageHistory as ChatMessageHistory
 8from langchain_core.runnables.history import RunnableWithMessageHistory
 9
10# RunnableWithMessageHistory 등은 LangGraph 권장으로 deprecated 처리됨 (2.0 전까지 동작)
11# langchain_core가 import 시 자체 경고 필터를 켜므로, import 이후에 꺼야 합니다
12warnings.filterwarnings("ignore", category=LangChainDeprecationWarning)
13
14load_dotenv()
15
16# 세션별 대화 기록 저장소 (실제 서비스에서는 Redis나 DB로 교체합니다)
17_store: dict[str, ChatMessageHistory] = {}
18
19def get_session_history(session_id: str) -> ChatMessageHistory:
20    """세션 ID별 기록 객체를 반환. 함수 시그니처만 유지하면 저장소는 무엇이든 됩니다."""
21    if session_id not in _store:
22        _store[session_id] = ChatMessageHistory()
23    return _store[session_id]
24
25model = ChatBedrockConverse(
26    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
27    region_name="us-east-1",
28)
29
30# 모델에 메모리를 결합: 기록 저장과 결합이 자동으로 처리됩니다
31with_memory = RunnableWithMessageHistory(model, get_session_history)
32
33# session_id로 사용자를 구분합니다
34config = {"configurable": {"session_id": "user_123"}}
35print("[1차]", with_memory.invoke("안녕, 내 이름은 김하늘이야.", config=config).content)
36print("[2차]", with_memory.invoke("내 이름이 뭐였지?", config=config).content)
37
38# 다른 세션은 격리됩니다: 같은 체인, 다른 기억
39other = {"configurable": {"session_id": "user_456"}}
40print("[다른 사용자]", with_memory.invoke("내 이름 알아?", config=other).content)

 1# tool_decision.py - 모델의 도구 호출 "결정" 확인
 2from dotenv import load_dotenv
 3from langchain_core.tools import tool
 4from langchain_aws import ChatBedrockConverse
 5
 6load_dotenv()
 7
 8@tool
 9def multiply(a: int, b: int) -> int:
10    """두 정수 a와 b를 입력받아 곱을 반환한다. 정확한 산술 계산이 필요할 때 사용."""
11    return a * b
12
13@tool
14def search_employee(name: str) -> str:
15    """삼성SDS 임직원 디렉터리에서 이름으로 직원 정보를 조회한다."""
16    # 교육용 더미 데이터입니다
17    fake_db = {
18        "김하늘": "김하늘 / 클라우드운영팀 / haneul.kim@samsungsds.example.com",
19        "박도윤": "박도윤 / 물류플랫폼팀 / doyun.park@samsungsds.example.com",
20    }
21    return fake_db.get(name, f"{name}을(를) 찾을 수 없습니다.")
22
23# bind_tools: 모델에게 "이런 도구들이 있다"고 알려줍니다
24llm = ChatBedrockConverse(
25    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
26    region_name="us-east-1",
27)
28llm_with_tools = llm.bind_tools([multiply, search_employee])
29
30queries = [
31    "1234567 곱하기 8901234 계산해줘.",              # 단일 도구
32    "김하늘 찾아서 이메일 알려주고 15 곱하기 25도.",  # 복합 (도구 2개)
33    "그냥 안녕이라고만 인사해줘.",                    # 도구 불필요
34]
35
36for q in queries:
37    response = llm_with_tools.invoke(q)
38    print(f"[질문] {q}")
39    if response.tool_calls:  # 모델이 호출하기로 "결정"한 도구 목록 (아직 실행 전)
40        for call in response.tool_calls:
41            print(f"  -> 호출 결정: {call['name']}({call['args']})")
42    else:
43        print(f"  -> 직접 응답: {response.content}")
44    print()

  1# tool_playground.py - 여러 도구를 붙여 "결정 -> 실행 -> 최종 답변"까지 테스트
  2# 추가 패키지: pip install ddgs tzdata
  3from datetime import datetime
  4from zoneinfo import ZoneInfo
  5
  6import requests
  7from ddgs import DDGS
  8from dotenv import load_dotenv
  9from langchain_aws import ChatBedrockConverse
 10from langchain_core.messages import HumanMessage, ToolMessage
 11from langchain_core.tools import tool
 12
 13load_dotenv()
 14TIMEOUT = 10  # 외부 API 대기 시간(초)
 15
 16
 17# --- 기존 도구 (tool_decision.py와 동일) ---
 18@tool
 19def multiply(a: int, b: int) -> int:
 20    """두 정수 a와 b를 입력받아 곱을 반환한다. 정확한 산술 계산이 필요할 때 사용."""
 21    return a * b
 22
 23
 24@tool
 25def search_employee(name: str) -> str:
 26    """삼성SDS 임직원 디렉터리에서 이름으로 직원 정보를 조회한다."""
 27    # 교육용 더미 데이터입니다
 28    fake_db = {
 29        "김하늘": "김하늘 / 클라우드운영팀 / haneul.kim@samsungsds.example.com",
 30        "박도윤": "박도윤 / 물류플랫폼팀 / doyun.park@samsungsds.example.com",
 31    }
 32    return fake_db.get(name, f"{name}을(를) 찾을 수 없습니다.")
 33
 34
 35# --- 새 도구 ---
 36@tool
 37def web_search(query: str, max_results: int = 3) -> str:
 38    """인터넷에서 최신 정보를 검색한다. 뉴스, 최근 사건, 모델이 모를 수 있는 사실을 확인할 때 사용."""
 39    try:
 40        results = DDGS().text(query, max_results=max_results)
 41    except Exception as e:
 42        return f"검색 실패: {e}"
 43    if not results:
 44        return "검색 결과가 없습니다."
 45    return "\n\n".join(f"- {r['title']}\n  {r['body']}\n  {r['href']}" for r in results)
 46
 47
 48@tool
 49def wikipedia_summary(title: str) -> str:
 50    """한국어 위키백과에서 문서 제목으로 요약을 가져온다. 인물, 개념, 용어의 정의를 찾을 때 사용."""
 51    url = "https://ko.wikipedia.org/api/rest_v1/page/summary/" + requests.utils.quote(title)
 52    try:
 53        r = requests.get(url, headers={"User-Agent": "sds-ax-practice/1.0"}, timeout=TIMEOUT)
 54    except requests.RequestException as e:
 55        return f"위키백과 조회 실패: {e}"
 56    if r.status_code == 404:
 57        return f"'{title}' 문서를 찾을 수 없습니다."
 58    return r.json().get("extract", "요약이 없습니다.")
 59
 60
 61WEATHER_CODES = {0: "맑음", 1: "대체로 맑음", 2: "구름 조금", 3: "흐림", 45: "안개", 48: "안개",
 62                 51: "이슬비", 53: "이슬비", 55: "이슬비", 61: "비", 63: "비", 65: "강한 비",
 63                 71: "눈", 73: "눈", 75: "강한 눈", 80: "소나기", 81: "소나기", 82: "강한 소나기",
 64                 95: "뇌우", 96: "뇌우", 99: "뇌우"}
 65
 66
 67@tool
 68def get_weather(city: str) -> str:
 69    """도시의 현재 날씨(기온, 날씨 상태, 습도, 풍속)를 조회한다. city는 반드시 영문 도시명으로 넣는다. 예: Seoul, Busan, Tokyo."""
 70    try:
 71        geo = requests.get("https://geocoding-api.open-meteo.com/v1/search",
 72                           params={"name": city, "count": 1, "language": "ko"}, timeout=TIMEOUT).json()
 73        if not geo.get("results"):
 74            return f"'{city}' 도시를 찾을 수 없습니다. 영문 도시명으로 다시 시도하세요."
 75        place = geo["results"][0]
 76        cur = requests.get("https://api.open-meteo.com/v1/forecast",
 77                           params={"latitude": place["latitude"], "longitude": place["longitude"],
 78                                   "current": "temperature_2m,relative_humidity_2m,wind_speed_10m,weather_code"},
 79                           timeout=TIMEOUT).json()["current"]
 80    except requests.RequestException as e:
 81        return f"날씨 조회 실패: {e}"
 82    state = WEATHER_CODES.get(cur["weather_code"], f"코드 {cur['weather_code']}")
 83    return (f"{place['name']}({place.get('country', '')}) 현재 날씨: {state}, "
 84            f"기온 {cur['temperature_2m']}°C, 습도 {cur['relative_humidity_2m']}%, 풍속 {cur['wind_speed_10m']}km/h")
 85
 86
 87@tool
 88def get_current_time(timezone: str = "Asia/Seoul") -> str:
 89    """지정한 시간대의 현재 날짜와 시각을 반환한다. timezone은 IANA 형식이다. 예: Asia/Seoul, America/New_York."""
 90    try:
 91        now = datetime.now(ZoneInfo(timezone))
 92    except Exception:
 93        return f"알 수 없는 시간대입니다: {timezone}"
 94    return now.strftime(f"%Y-%m-%d %H:%M:%S ({timezone})")
 95
 96
 97@tool
 98def convert_currency(amount: float, from_currency: str, to_currency: str) -> str:
 99    """최신 환율로 금액을 환전 계산한다. 통화는 ISO 코드로 넣는다. 예: USD, KRW, JPY, EUR."""
100    try:
101        data = requests.get(f"https://open.er-api.com/v6/latest/{from_currency.upper()}", timeout=TIMEOUT).json()
102    except requests.RequestException as e:
103        return f"환율 조회 실패: {e}"
104    rate = data.get("rates", {}).get(to_currency.upper())
105    if data.get("result") != "success" or rate is None:
106        return f"지원하지 않는 통화입니다: {from_currency} -> {to_currency}"
107    return (f"{amount:,.2f} {from_currency.upper()} = {amount * rate:,.2f} {to_currency.upper()} "
108            f"(환율 {rate:,.4f}, 기준 {data.get('time_last_update_utc', '알 수 없음')})")
109
110
111TOOLS = [multiply, search_employee, web_search, wikipedia_summary,
112         get_weather, get_current_time, convert_currency]
113TOOLS_BY_NAME = {t.name: t for t in TOOLS}
114
115llm = ChatBedrockConverse(
116    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
117    region_name="us-east-1",
118)
119llm_with_tools = llm.bind_tools(TOOLS)
120
121
122def run(question: str, max_turns: int = 5) -> dict:
123    """모델이 도구를 고르고, 우리가 실행하고, 결과를 돌려주는 과정을 반복합니다."""
124    print(f"[질문] {question}")
125    messages = [HumanMessage(question)]
126    usage_sum = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "calls": 0}
127    for turn in range(1, max_turns + 1):
128        response = llm_with_tools.invoke(messages)
129        messages.append(response)
130
131        # 모델 호출 1회의 토큰 사용량 (도구 실행 자체는 토큰을 쓰지 않습니다)
132        usage = response.usage_metadata or {}
133        for key in ("input_tokens", "output_tokens", "total_tokens"):
134            usage_sum[key] += usage.get(key, 0)
135        usage_sum["calls"] += 1
136        print(f"  [토큰 {turn}회차] 입력 {usage.get('input_tokens', 0):,} / "
137              f"출력 {usage.get('output_tokens', 0):,} / 합계 {usage.get('total_tokens', 0):,}")
138
139        if not response.tool_calls:  # 더 부를 도구가 없으면 최종 답변
140            print(f"  [최종 답변] {response.text}")
141            break
142        for call in response.tool_calls:
143            print(f"  -> 호출 결정: {call['name']}({call['args']})")
144            result = TOOLS_BY_NAME[call["name"]].invoke(call["args"])
145            print(f"     결과: {str(result)[:150]}")
146            messages.append(ToolMessage(content=str(result), tool_call_id=call["id"]))
147    else:
148        print("  [중단] 최대 반복 횟수에 도달했습니다.")
149    print(f"  [토큰 질문 합계] 모델 호출 {usage_sum['calls']}회 / 입력 {usage_sum['input_tokens']:,} / "
150          f"출력 {usage_sum['output_tokens']:,} / 합계 {usage_sum['total_tokens']:,}")
151    print()
152    return usage_sum
153
154
155queries = [
156    "지금 서울 날씨 어때?",
157    "100달러는 원화로 얼마야? 그리고 지금 뉴욕은 몇 시야?",
158    "랭체인(LangChain)이 뭔지 위키백과 기준으로 알려줘.",
159    "삼성SDS 최근 뉴스 검색해서 두 줄로 요약해줘.",
160    "박도윤 이메일 알려주고, 부산 날씨도 알려줘.",
161    "그냥 안녕이라고만 인사해줘.",  # 도구 불필요
162]
163
164if __name__ == "__main__":
165    grand = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "calls": 0}
166    for q in queries:
167        for key, value in run(q).items():
168            grand[key] += value
169    print("=" * 60)
170    print(f"[전체 토큰 합계] 질문 {len(queries)}개 / 모델 호출 {grand['calls']}회")
171    print(f"  입력 {grand['input_tokens']:,} / 출력 {grand['output_tokens']:,} / 합계 {grand['total_tokens']:,}")