오늘은 어제 수업에 이어서 RAG TEST 3번째 파트를 마무리 하고,
Agent 개발 컨셉 수업과
mcp 기초적 실습
langchain 실습
LLM, RAG 구성
LangChain 구성 등의 수업을 진행하였다.
우리가 만드는게 AI Agent라니 나 3달밖에 안했는데 믿기지가않는데 이게말이되나 이제 당장 이번주부터 미니프로젝트로 ai agent 만들어야된다 근데 내 주제로 하고싶긴 한데 내 주제 완전 좋을줄 알았는데 다른 사람들거 얘기 들어보니까 그것도 좋을거같아서 고민이다 내꺼로 하고싶은데
- 세 가지 주제 중 두가지로 골랐는데 그 중 하나가 내 걸로 됐다.
물론 주제가 신박해서, 잘 해서는 아닌거같고 그냥 셋이 준비했는데 한 분은 프로젝트 하눈데 의의를 두고 나머지 두개로 정했다 ㅎ
아무튼 복습
rag test 3
이제 복습을 시작하려던 와중 집에 와서 깃 클론 해와서 다시 실행하니까 안되는 오류가 있었다.
원인은 내가 쓰는 aws 리전에 수업시간에 사용한 임베딩 모델을 지원하지 않는것이었다.
그래서 해결방법을 알아봤는데 런던에서 사용가능한 임베딩 모델로 바꾸면 됐다.
그런데도 안되어서 이유를 더 찾아보니
rag_test2에서 만든 폴더 hp_Story 인덱싱부터 이전 임베딩 모델로 만들었기 때문에
호환이 되지 않아서 그랬던 것이었다. 이제 수정 완료
'''
- 저장된 백터 디비 로드
- llm 이용한 rag 사용 => 검색 증강 => llm 모르는 내용을 데아터를 전달하여 추론
- 질의 -> 검색 -> 결과획득 -> 프럼프트(질의 + 검색결과) -> 추론요청 -> 응답 -> 확인
- 랭체인과 연동되어서 체인 구성(파이프라인)
'''
# 1. 모듈 가져오기
from langchain_community.vectorstores import FAISS
from langchain_aws import BedrockEmbeddings
import boto3
from dotenv import load_dotenv
import os
from langchain_aws import ChatBedrock
from langchain_core.prompts import ChatPromptTemplate
# RunnablePassthrough 질문을 검색하면서 동시에 사용자 질문을 세팅함
from langchain_core.runnables import RunnablePassthrough
# StrOutputParser llm의 응답을 파싱하여 문자열만 추출
from langchain_core.output_parsers import StrOutputParser
load_dotenv()
AWS_REGION = os.getenv('AWS_REGION')
print( AWS_REGION )
# 임베딩모델(토크나이저 역활)
tokenizer = BedrockEmbeddings( model_id = "cohere.embed-multilingual-v3",
region_name = os.getenv('AWS_REGION') )
# **디비로드**
db = FAISS.load_local('hp_story', tokenizer, allow_dangerous_deserialization=True)
# 검색(유사도 테스트, 유사도가 가장 높은 문장 1개만 출력)-마지막만 출력(20글자)
print( db.similarity_search("해리포터의 친구")[0].page_content[-20:])
# 1. aws bedrock 클라이언트 구성
bedrock_client = boto3.client(
service_name = 'bedrock-runtime',
region_name = AWS_REGION
)
# 2. llm 생성
llm = ChatBedrock(
client = bedrock_client,
model_id = 'openai.gpt-oss-120b-1:0', # os.getenv('BEDROCK_MODEL_ID'), 구글은 ChatBedrock 미지원
model_kwargs = {
"temperature": 0.7,
"max_tokens" : 500
}
)
# 3. prompt 구성
prompt = ChatPromptTemplate.from_template('''
다음의 제공된 context(문맥, 참고)을 사용하여 질문에 답변해 주세요.
만약, 문맥에서 답을 찾을 수 없다면, "잘 모르겠다"고 대답 하세요.
<context>
{context}
</context>
질문: {user_input}
''')
# 4. 체인 구성 : 초기스타일 <-> 최근 스타일:파이프 연산자(|) 사용하여 LCEL(Langchain Expression Language)
# 4-1. 리트리버 생성 : DB에서 유사도 높은 문서를 찾아온다 -> top 3 제한
retriever = db.as_retriever(search_kwargs={"k":3}) # 상위 3개 문서 참조
# 4-2. 문서 결합 체인 : 검색된 문서들을 프럼프트의 {context} 세팅한다
def format_docs(docs):
# 탑 3개 검색 => 한문장으로 결합
return "\n\n".join(doc.page_content for doc in docs)
# 4-3. 최종 RAG 체인 구성 : 질문->검색->프럼프트 결합->LLM 질의->답변획득
rag_chain = (
{"context": retriever | format_docs, "user_input": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
# --- 8. 실행 ---
query = "해리포터의 적은?"
print(f"질문: {query}\n")
response = rag_chain.invoke(query)
print("=== AI 답변 ===" )
print(response)이는 이전에 RAG_TEST2에서 만든 DB를 가져와서 실제로 RAG를 적용한 LLM 모델을 테스트하는 파일이다.
DB를 만들때 사용한 임베딩 모델을 그대로 가져와서
디비를 로드한다, 그리고 BEDROCK 클라이언트와 LLM모델을 생성하고 프롬프트를 세팅한다.
그리고 질문 -> 검색 -> 프롬프트 적용 -> LLM 질의 -> 답변 획득의 RAG체인을 구성하고
마지막으로 RAG_CHAIN을 INVOKE 해주면 답변이 나오는 것이다.
질문: 해리포터의 적은?
=== AI 답변 ===
<reasoning>We need to answer based on provided context. The question: "해리포터의 적은?" Means "Who are Harry Potter's enemies?" In context, we have mentions: "볼드모트의 부하" (Voldemort's minion), "리키 콜드런에서 위즐리 부부의 대화를 엿들은 해리는 볼드모트의 부하 로, 해리의 부모를 배신하고 두 사람을 죽음으로 내몬 인물이며 아즈카반을 탈옥한 시리우스 블랙이 자신의 목숨을 노리고 있다는 것을 알게 된다." Also mention "디멘터" as enemies. Also "볼드모트 경과 그의 추종자, 죽음을 먹는 자들" are enemies. So answer: Voldemort, Death Eaters, Dementors, maybe Sirius Black (but he is an ally). The context says Sirius is trying to kill him? Actually in canon, Sirius is his godfather, but the context says "시리우스 블랙이 자신의 목숨을 노리고 있다". So within this weird context, they consider Sirius as an enemy. Also "볼드모트의 부하" is a specific person (maybe Peter Pettigrew?). The question is generic: "해리포터의 적은?" We can answer: 주요 적은 볼드모트와 그의 추종자들인 죽음을 먹는 자들, 디멘터, 그리고 볼드모트의 부하인 피터 페티그루 등이다. Also mention "시리우스 블랙" as perceived enemy in this context. Provide answer in Korean. If not found, say "잘 모르겠다". But we have enough.</reasoning>해리 포터의 주요 적은 다음과 같습니다.
- **볼드모트(어둠의 마왕)** – 해리의 가장 큰 적이자 전체 이야기를 뒤흔드는 존재.
- **죽음을 먹는 자들(볼드모트의 추종자)** – 볼드모트에 맞서 싸우는 조직으로, 해리와 친구 들을 위협한다.
- **디멘터** – 어둠의 감옥 아즈카반에서 일하는 사령관들로, 해리에게 정신적인 고통을 주고 퀴디치 경기 중에도
PS D:\project\T아카데미\python\llm_langchain_bedrock\test>결과물이다.
LLM이 모르느 내용이라도 RAG를 통해서 DB에 저장된 토큰화된 데이터를 읽어서 답변을 추출하는 것이다.
langgraph_test1
- **LangGraph**
- 특징 : **Stateful** Graph 구조. **상태(state)**를 기억하고 노드 간을 이동
- 강점 : 순환(cycle)구조 가능. LLM의 결과물에 에러(불만족등)가 발생하면 다시 수정(이전)단계로 진입. 복잡한 에이전트 개발에 적합(필수적)
- 기타 : **LangChain기반으로 만드어진 기술** LangGraph임 대체가 아닌 보완으로 이해
- 상호 보완적 사용
- 적합한 프로젝트 : 생각하는(자율형) 에이전트 (검색해 보고 부족하다고 느끼면 다시 검색 진행하는 등), 멀티 에이전트(A2A, 검색에이전트, 작문에이전트, 평가에이전트..) 협업시, 사업이 중간에 개입하여 승인/거절등 행위가 사용될때(Human in the loop)
langgraph 는 langchain에서 업그레이드 된 기술로
결과에 에러(불만족한 결과물)이 나오면 각 노드를 다시 이동하며 적합한 결과물을 도출해내는 과정이다.
# 1. 모듈 가져오기
from langgraph.graph import StateGraph, END
from typing import TypedDict
# 2. 상태 정의 (데이터 담을 그릇 -> 차후 프롬프트 들어가는 재료 등 포함)
# 클래스형, 클래스의 슈퍼클래스로 TypedDict 를 통상 사용 (Basemodel느낌)
# TypedDict상속받은것은 fastapi의 pydantic과 거의 유사성을 가짐
class CustomState(TypedDict):
msg:str
# 3. 노드 준비 (작업 내용 -> tool(rag등..)로 이해 -> mcp 연계)
# 현재는 툴 수준으로 설정하지는 않고, 단순 함수로 구성
def add_prefix(state:CustomState):
# 기존 상태값의 앞에 특정 내용을 추가함 => 상태값 업데이트
return {'msg':'헬로'+state['msg']}
# 기존 상태값의 뒤에 특정 내용을 추가함 => 상태값 업데이트
def add_surfix(state:CustomState):
return {'msg': state['msg'] + '!!'}
# 4. 그래프 연결
# 4-1. 그래프 생성 (구조적 껍데기)
workflow = StateGraph(CustomState) # CustomState의 형태로 상태가 관리되는 상태그래프 생성
# 4-2. 노드(tool) 추가 -> 서클형 -> 시작과 끝을 모름
workflow.add_node("S1" , add_prefix)
workflow.add_node("S2" , add_surfix)
# 4-3. 시작점 설정
workflow.set_entry_point('S1')
# 4-4. 작업 순서를 설정(방향성)
workflow.add_edge('S1','S2') # S1이 끝나면 S2로 진입
# 4-5. 끝나는 방향성 설정
workflow.add_edge('S2',END) # S2가 끝나면, 종료
# 4-6. 컴파일 -> 수행 가능 단위 구성 -> 완성
app = workflow.compile()
# 5. 실행 -> 그래프 기반 사용자 질의 (데이터 포함) -> 자율형을 그래프를 돌면서 해결
# dict 형태여애 한다 -> 상태는 TypedDict를 상속받음 -> 키는 msg -> msg:str
# 상태에 대한 형태를 정의했으므로 그에 맞게 입력
res = app.invoke({"msg":"월드"})
print(res)
'''
"][=현재 코드는 invoke를 통해서 상태값이 초기 세팅이 되면
입력 :
{"msg":"월드"}
작동 :
S1 노드에서 시작되어서 헬로가 추가됨 ( 상태 업데이트 )
S2 노드에 진입되어서 !! 추가됨 ( 상태 업데이트 )
끝나면 -> 그래프 순환작업 완료 -> 응답 -> 최종 상태값을 반환
응답 :
{'msg': '헬로월드!!'}
추후 에이전트 판단 LLM에 질의할 정도의 프롬프트가 아니면 보충(tool을 통해서)하는 방식 이해
'''설명을 읽으면 다 알지만 간단히
add_prefix는 {msg:월드}인 dict 구조에 앞에 헬로를 붙여주고
add_surfix는 dict 구조의 월드 뒤에 !! 를 붙여주는 것이다.
결과물은 헬로월드!! 로 나온다. 그리고 이번 파일은 노드(tool)의 이해를 돕기 위한 예제이다.
서클형으로 우리가 방향성을 정해서 그 구조에 맞게 툴이 돌아가고 엔드포인트를 정해주면 거기서 멈추는것이다.
실제로는 llm이 어디서 멈출 지를 판단해서 멈춘다.
langgraph_test2
본격적으로 langgraph 의 구조를 알아보는 예제이다.
# 1. 모듈 가져오기
from langgraph.graph import StateGraph, END, MessagesState, START
from typing import TypedDict
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langchain_aws import ChatBedrockConverse, ChatBedrock
from langgraph.prebuilt import ToolNode, tools_condition # 툴을 노드로 변환, 상황(조건)에 따라 툴 적용함수
from dotenv import load_dotenv
import os
load_dotenv()
import boto3
# 툴
@tool # 데코레이터, LLM이 알 수 있는(이해할 수 있는) 형식(포멧)으로 자동 변환됨
def multiply(a:int, b:int)->int:
''' 두 수를 곱한 후 반환 '''
print(f' [Tool 실행] {a} X {b} 계산중...')
return a * b
# llm=ChatBedrockConverse( model=os.getenv('BEDROCK_MODEL_ID'),
# region_name=os.getenv('AWS_REGION'))
# 모델 id는 anthropic.claude-3-sonnet-20240229-v1:0 사용
llm=ChatBedrock(model=os.getenv('BEDROCK_MODEL_ID'),
client=boto3.client('bedrock-runtime',region_name=os.getenv('AWS_REGION')))
tools = [multiply]
llm_with_tools = llm.bind_tools(tools) # llm에게 이런 툴을 사용할 수 있다라는 것을 알림(등록)
# 노드 신규 구성 -> 함수 정의
def chatbot_node(state:MessagesState):
# 계속된 대화내용이 `누적`되어서 LLM에게 전달하여 추론 행위가 진행됨
# 누적 -> 히스토리 -> 대화 내용을 계속해서 LLM에게 전달하여 대화가 이어지게 됨
return {"messages":[llm_with_tools.invoke(state['messages'])]}
# 그래프 구성상 상태 -> 커스텀하지 않고 MessagesState 상태값의 멤버는 'messages'
# 그래프 생성
workflow = StateGraph(MessagesState)
# 노드 추가
workflow.add_node("chatbot" , chatbot_node) # 대화 내용을 보고 생각 -> 뇌 담당 노드
workflow.add_node("tools" , ToolNode(tools)) # 툴을 노드로 변환하여 그래프에 추가 -> 행동(수단)(곱해라) -> 행동(수단)담당 노드
# 시작점
workflow.add_edge(START,'chatbot') # 서비스 가동 -=> 가장 먼저 챗봇 가동됨
# 조건에 따라 행동을 다르게 수행
workflow.add_conditional_edges( # 조건부
'chatbot', # 이전 노드가 텍스트를 응답했으면 끝
tools_condition) # 이전 노드가 도구가 필요하다고 응답하면 -> 도구 노드로 이동
# 도구 사용 -> 결과 획득(추론을 위한 보충자료 ) 획득 -> 챗봇으로 전달 -> 다시 추론행위
workflow.add_edge('tools', 'chatbot')
# 사이클 cast 2종류
# 질의 -> chatbot_node -> llm 호출 -> 응답 -> end
# 질의 -> chatbot_node -> llm 호출 -> 부족함 도구 사용 필요 -> 툴 -> 툴 사용 -> 결과
# chatbot_node -> llm 호출 -> 응답 -> end
app = workflow.compile()
# 테스트
if __name__ == "__main__":
print('Agent 시작, 종료시 q 입력')
while True:
# 1. 질의 획득
user_input = input('\n사용자:')
# 2. 탈출 코드
if user_input.lower() == 'q':break
# 3. 프롬프트 구성 (단순하게)
prompt = {'messages':[HumanMessage(content=user_input)]} # 상태와 동일 행태로 구성
# 4. 그래프 작동(invoke: 동기식 요청 - 응답, stream: 비동기식, 실시간 중계, 스트리밍)
# 휘발성으로 이전 대화 내용이 모두 초기화됨
for evt in app.stream(prompt, stream_mode='values'):
msg = evt['messages'][-1] # 가장 최근에 추가된 내용 -> 실시간 응답
# 5. 출력 실시간 출력 x, 마지막값 한번에 출력)
print('Agent', msg.content)
결과값 1
결과값 2
애로사항
이는 langgraph에서 llm이 직접 판단하여 도구를 사용할 지 말지, 그 과정을 알아보는 코드이다.
먼저 기본 구성자체는 같으나 tools에 multiply 기능을 넣었다. 곱셈을 해주는 기능이다.
1에서는 llm을 불러오지 않았고 방향성만 설명하였고, 2에서는 llm을 불러왔다.
calude를 사용했다.
그리고 챗봇 노드를 구성하여 함수를 정의하였다.
챗봇 노드, 툴 노드를 설정하였고 툴을 노드로 변환하여 그래프에 추가하였다.
그 노드들을 돌며 llm이 판단하여 결과값을 줄 것이고, 그게 랭그래프다.
이 떄 조건부 엣지를 주어서 도구를 사용하지 않아도 되면 도구 사용이 출력에 안 나온다 <- 결과값 1
도구를 사용해야 할거같으면 도구를 사용한다 <- 결과값 2
그리고 프롬프트와 질의 획득하는 코드를 넣어서 결과값을 받았다.
그런데 애로사항을 보면, 방금 전 결과를 기억하지 못하고 이상하게 답변한다, 질문에 답을 하는 순간 모든 정보가 휘발된다.
이를 해결하기 위해서 챕터 3 나가신다
langgraph_test3
from langgraph.graph import StateGraph, END, MessagesState, START
from typing import TypedDict
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langchain_aws import ChatBedrockConverse, ChatBedrock
from langgraph.prebuilt import ToolNode, tools_condition # 툴을 노드로 변환, 상황(조건)에 따라 툴 적용함수
from dotenv import load_dotenv
import os
load_dotenv()
import boto3
# 1. 모듈 가져오기 (메모리 저장 (단기기억, 프로그램 종료되면 삭제됨))
from langgraph.checkpoint.memory import MemorySaver
# 2. 메모리 생성-> RAM에 공간을 할당함 -> 실제는 물리적 디비에 저장 (백터디비, RDB등)
memory = MemorySaver()
@tool
def multiply(a:int, b:int)->int:
''' 두 수를 곱한 후 반환 '''
print(f' [Tool 실행] {a} X {b} 계산중...')
return a * b
llm=ChatBedrock(model=os.getenv('BEDROCK_MODEL_ID'),
client=boto3.client('bedrock-runtime',region_name=os.getenv('AWS_REGION')))
tools = [multiply]
llm_with_tools = llm.bind_tools(tools)
def chatbot_node(state:MessagesState):
return {"messages":[llm_with_tools.invoke(state['messages'])]}
workflow = StateGraph(MessagesState)
workflow.add_node("chatbot" , chatbot_node)
workflow.add_node("tools" , ToolNode(tools))
workflow.add_edge(START,'chatbot')
workflow.add_conditional_edges(
'chatbot',
tools_condition)
workflow.add_edge('tools', 'chatbot')
# 랭그래프 생성시 컴파일 옵션으로 단기기억 공간 제공
# 실행될 때마다 memory에 자동 저장됨
app = workflow.compile(checkpointer=memory)
if __name__ == "__main__":
print('Agent 시작, 종료시 q 입력')
# 메모리 저장 시 설정값
'''
현재는 id를 고정값으로 구성 -> id가 같으면 같은 대화방/채팅으로 인식하고 기억 설정
로드 -> 병합 -> 실행 단계로 메모리에 기억 => [과거기록 + 새 질문]
'''
config = {"configurable":{"thread_id":"user-1"}} # 사용자별 아이디로 관리 -> user-1
while True:
user_input = input('\n사용자:')
if user_input.lower() == 'q':break
prompt = {'messages':[HumanMessage(content=user_input)]}
# 스트림 진행시 설정값 세팅
for evt in app.stream(prompt, stream_mode='values', config=config):
msg = evt['messages'][-1]
print('Agent', msg.content)
이 코드는 위의 2번 코드와 다른점은 단기기억 공간 제공밖에 없다.
app = workflow.compile(checkpointer=memory)
요기
그리고 id를 지정해줘서 id가 같으면 같은 사람으로 인식하고 그 결과값을 이어간다.
이전의 내용을 기억해서 대답을 이어가는 모습이다.
오늘의 수업은 여기까지였다.
크리스마스 주간이고 미니프로젝트도 들어가서 토요일 밤에야 급하게 복습에 들어간다