[KT AIVLE SCHOOL] AGENT

AI AGENT 이해

 

 

TOOL부분에서부턴 이해가 거의 안 되네요 나중에 차근차근 수정하면서 알아가겠습니다

 

 

AI AGENT 개요

 

 

 

AI AGENT?

• 목표가 있고, 그 목표를 달성하도록 만들어놓은 시스템
• LLM을 중심으로 다양한 외부 도구를 연결하여 구축
• Tool사용 능력+ 메모리 사용 자율성+gpt = 문제를 스스로 해결하는 시스템

 

ex. 문서요약& 질의응답 에이전트 , 쇼핑 추천 에이전트, 코딩 도우미 등

 

 

 

LangGraph 기반 Agent 시스템 1

 

💥 LangGraph 

AI Agent 간의 협업을 그래프(흐름도 생각하기) 기반으로 설계하는 프레임워크

 

 

💥용어정리

노드 : 특정 작업을 수행하는 함수.

엣지 : 연결

조건부엣지 : 조건에 따라서 노드 간의 분기 처리 가능하게 함

스테이트 : task의 현재 결과물 틀 ( task1이 끝날 때의 결과물 , 즉 task2의 인풋)

 

 

 

💥그래프 유형

 

1. 간단 그래프

 

• State 정의

from typing_extensions import TypedDict
# 그래프 전체에서 주고받는 데이터 구조를 정의
class State(TypedDict):
text: str
extra_field: int

 

틀 정의

 

 

 

• node 정의

def node_1(state: State): # State는 그 state가 어떤 키와 값을 갖는지 미리 정의한 타입
# 입력: 상태 (state) → text 필드에 문자열을 덧붙임
# extra_field은 10으로 덮어씀
return {"text": state['text'] + "(텍스트 추가됨)", "extra_field": 10}

 

 

2.Routing

1

 

 

• State

class State(TypedDict):
    number: int
    result: str

 

 

 

• Node

 두 개의 노드로 갈라지는 노드는 check_parity로 선언한다

# 노드: 짝/홀수 판별
def check_parity(state: State):
    print(f"check_parity: 입력된 숫자 = {state['number']}")
    return state # 입력만 받고, 분기는 conditional_edge에서 처리

 

#짝수 노드
def even_node(state: State):
    print("짝수입니다!")
    state["result"] = "짝수입니다!"
    return state
# 홀수 노드
def odd_node(state: State):
    print("홀수입니다!")
    state["result"] = "홀수입니다!"
    return state

 

 

• 그래프 구조

# 그래프 생성
builder = StateGraph(State)

# 노드 등록
builder.add_node("check_parity", check_parity)
builder.add_node("even_node", even_node)
builder.add_node("odd_node", odd_node)

# 조건 분기 연결
def parity_condition(state: State):
    return "even" if state["number"] % 2 == 0 else "odd"
    
builder.add_conditional_edges("check_parity", parity_condition,
    							{"even": "even_node", "odd": "odd_node"})

# 나머지 연결
builder.add_edge(START, "check_parity")
builder.add_edge("even_node", END)
builder.add_edge("odd_node", END)

# 그래프 컴파일
graph = builder.compile()

 

 

 

 

3.Reflection

LLM이 스스로 추론 과정을 반복하며 피드백 생성하는 매커니즘

 

• state정의

class State(TypedDict):
input: str
summary: str
is_satisfied: bool
log: list[str]

 

 

• node정의

# 요약 노드: 간단히 결과 요약 + 만족 여부 결정
def summary_node(state: State):
    state["summary"] += " → 요약됨"
    state["log"].append("요약 수행")
    # 만족 여부 판단 (2회 이상 요약되면 만족했다고 가정)
    if state["summary"].count("요약됨") >= 2:
        state["is_satisfied"] = True
    else: state["is_satisfied"] = False
        print(state)
    return state
# 반추 노드: 추가 아이디어 도출 시도 (다시 생각해보기)
def reflect_node(state: State):
    state["log"].append("반추 수행")
    state["summary"] += " → 다시 생각해봄"
    print(state)
    return state

 

 

• 그래프 구조

# 그래프 구성
builder = StateGraph(State)
builder.add_node("summarize", summary_node)
builder.add_node("reflect", reflect_node)

# 흐름 정의
builder.add_edge(START, "summarize")

# 조건 분기: 만족 여부에 따라 흐름 결정
def check_satisfaction(state: State):
    return "satisfied" if state["is_satisfied"] else "retry"
builder.add_conditional_edges("summarize", check_satisfaction,
    {"satisfied": END, "retry": "reflect"})

# 반추 후 → 요약 다시 시도 (루프)
builder.add_edge("reflect", "summarize")
# 컴파일
graph = builder.compile()

 

 

• 그래프 실행

# 실행
input = {"input": "오늘 하루 요약해줘",
        "summary": "",
        "is_satisfied": False,
        "log": [] }
        
result = graph.invoke(input)
print("최종 상태:")
print(result)

 

 

 

LangGraph 기반 Agent 시스템 2 :Tools

 

💥Agent의 일반적인 구조

LLM은 사용자의 입력을 받아 응답을 생성한다.

필요시, 외부에서 정보를 검색하거나, 툴을 이용하거나 ,메모리를 읽고쓰며 결과를 종합해서 응답을 생성한다 .

 

 

 

 

💥tools

에이전트가 특정 작업을 수행하기 위해 호출할 수 있는 외부 기능

ex.계산기, 웹 검색, db조회, 시간 확인 등

 

 

 

 

💥간단한 CHATBOT 만들기

 

state 정의

• langGraph에서 llm 메시지 히스토리를 자동으로 관리하기 위한 방식
• 목록이 업데이트 될 때 덮어쓰지 않고 새롭게 추가

class State(TypedDict):
	messages: Annotated[list, add_messages]

 

 

node정의

• 언어 모델을 gpt-4o로 설정
• 응답을 리스트로 감싸서 message에 넣음

llm = ChatOpenAI(model="gpt-4o-mini")
def chatbot(state: State):
    result = llm.invoke(state["messages"])
    return {"messages": result} # 응답 "messages"에 넣어

 

 

그래프정의

builder = StateGraph(State)

builder.add_node("chatbot", chatbot)

builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

graph = builder.compile()

 

 

그래프 실행

result = graph.invoke({"messages": [HumanMessage("안녕")]})

 

 

 

💥custom tool을 사용한 agent

 

 

 

 

 

• tool 준비

@tool , ''' ''' 으로 tool설명

@tool
def calculator_tool(expression: str) -> str: # type hint : 함수의 출력 type은 str
    '''수식을 입력 받아 계산하는 도구'''
    try:
        result = eval(expression) # Python 내장 함수 eval()을 사용해서 문자열로 된 식을 계산
        return f"계산 결과: {result}"
    except Exception as e:
    	return f"오류 발생: {str(e)}"
        
@tool
def get_weather(location: str):
    """현재 날씨 정보를 반환합니다."""
    # 입력된 지역이 "서울"인 경우에 해당하는 날씨 정보를 반환합니다.
    if location in ["서울"]:
    	return "현재 온도는 20도이며 맑고 화창한 날씨입니다."
    else:
        # "서울"이 아닌 경우, 다른 날씨 정보를 반환합니다.
        return "현재 온도는 10도이며 조금 쌀쌀합니다."

 

 

 

 

• tool 준비2

#Tool Node로 묶어줄 때, 리스트 형태로 tool들을 함께 제공합니다.
tools = [get_weather, calculator_tool]
tool_node = ToolNode(tools)

 

 

 

• llm 준비

bind_toolsㅡ>이거로 메시지 생성

#bind_tools() 함수로 LLM에게 어떤 tool들이 있는지 인지시켜 도구를 활용하도록 함
from langchain_openai import ChatOpenAI

llm_with_tools = ChatOpenAI(model="gpt-4o-mini", temperature=0
                            ).bind_tools(tools)

 

 

• 그래프 만들기

# 모델 호출 함수 (GPT 호출)
def call_model(state: State):
    messages = state[“messages”]
    response = llm_with_tools.invoke(messages)
    return {“messages”: [response]}
    
# 조건 분기 함수: 툴 호출이 필요한가?
def should_continue(state: State) -> Literal["tools", END]:
    messages = state["messages"]
    last_message = messages[-1] # 가장 최근 message
    if last_message.tool_calls:
    	return "tools"
    return END

 

builder = StateGraph(State)

# 노드 추가
builder.add_node("call_model", call_model)
builder.add_node("tools", tool_node)

# 연결
builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", should_continue,
# {"tools": "tools", END: END}
)
builder.add_edge("tools", "call_model") # 루프 연결

# 컴파일
graph = builder.compile()

 

 

 

 

 

 

LangGraph 기반 Agent 시스템 3 :Memory

 

💥Memory

 

AI Agent가 이전 상태나 대화 기록을 유지하고 재사용할 수 있도록 하는 기능

 

💥MemorySaver

 

LangGraph에서 제공하는 체크포인트 저장소. 그래프 실행 중 생성되는 상태를 자동으로 저장하고 필요 시 복원할 수 있게 해주는 메모리 기반 저장소

 

💥MemorySaver 사용하기

 

• 메모리준비

from langgraph.checkpoint.memory import MemorySaver # MemorySaver 모듈 가져오기
memory = MemorySaver()

 

 

• 컴파일

(*컴파일은 프로그래밍에서 인간이 이해할 수 있는 소스 코드를 기계가 이해할 수 있는 기계어로 변환하는 과정)

# 체크포인터와 함께 그래프 컴파일
agent_with_memory = builder.compile(checkpointer = memory)

 

 

• 스레드 지정

# 대화 흐름을 구분하기 위한 thread_id 설정
thread_1 = {"configurable": {"thread_id": "1"}}

 

 

 

• 실행

messages = [HumanMessage(content="3과 4를 더하라")]
messages = agent_with_memory.invoke({"messages": messages}, thread_1)

messages = [HumanMessage(content="거기에 2를 곱하라.")]
messages = agent_with_memory.invoke({"messages": messages}, thread_1)

 

 

 

 

 

 

💥MemorySaver 활용하기

 

 

1. 최근 메시지만 기억

 

노드를 정의할 때 다음과 같은 간단한 함수로 구현 가능. gpt에게 최근 2개 메시지만 넘겨서 대화 문맥을 간결하게 유지하고, 불필요한 과거 메시지 누적을 피하는 구조

def filter_messages(messages: list):
    # 최근 2개 메시지만 리턴하는 필터 함수 생성
    return messages[-2:]
    
def chatbot(state: State):
    messages = filter_messages(state["messages"])
    result = llm.invoke(messages)
    return {"messages": [result]}

 

 

 

2. 오래된 메시지 요약

 

a. 처음6개 대화 ㅡ> END

b. 7번째 입력 ㅡ> 요약생성, 오래된 메시지 삭제

c. 또 6개 쌓이면 ㅡ>요약 2 생성, 기존메시지(요약1+새메시지로 요약 이어쓰기) , 오래된 메시지 삭제

 

 

 

 

💥Long-term Memory : SqliteSaver

 

 

 

 

LangGraph 기반 Agent 시스템 3 :RAG

 

 

💥간단한 RAG

 

VECTOR DB

# CSV 파일 로드
csv_path = "sample.csv"
csv_loader = CSVLoader(file_path= path+csv_path)
documents_csv = csv_loader.load()

# 벡터 DB 정의
embedding = OpenAIEmbeddings(model="text-embedding-ada-002")
vectorstore = Chroma.from_documents(documents_csv, embedding,
									persist_directory= path+ "chroma_db")	

retriever = vectorstore.as_retriever()

 

 

 

노드정의

llm = ChatOpenAI(model="gpt-4o-mini")
rag_chain = RetrievalQA.from_chain_type(llm=llm, retriever=retriever)

def rag_node(state: MessagesState):
    user_msg = [m.content for m in state["messages"] if isinstance(m, HumanMessage)][-1]
    answer = rag_chain.invoke({"query": user_msg})
    return {"messages": [AIMessage(content=answer["result"])]}

 

 

 

💥Vector DB를 Tool로 사용

 

• gpt가 스스로 판단해서 검색할지 말지를 결정 
• 구현방식 : VectorDB를 LangGraph의 tool로 명시적으로 분리
• LLM호출과 retriever을 별도 노드로 구분

 

 

구축절차

1.VectorDB준비

2.Tool 정의

@tool
def vectordb_search(query: str) -> str:
    """내장 문서로부터 관련 정보를 검색합니다.""" # docstring 꼭 포함되어야 함!
    # 유사도 기준으로 상위 k개 문서를 벡터스토어에서 검색
    docs = retriever.get_relevant_documents(query)
    # 검색된 문서 내용들을 줄바꿈으로 구분하여 하나의 문자열로 반환
    return "\n\n".join(d.page_content for d in docs)
    
# tool 리스트
tools = [vectordb_search]

# 툴 실행 노드 (LangGraph 제공)
tool_node = ToolNode(tools)

 

3. LLM+Tool 바인딩

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)

 

 

4. 노드준비 & 분기 함수

# GPT 호출 노드
def call_model(state: State):
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}
    
# 툴 사용 여부 판단
def should_continue(state: State) -> Literal["tools", END]:
    last = state["messages"][-1] # 대화의 가장 마지막 객체(가장 최근 대화)
    # last에서 tool_calls 내용이 포함되면, 'tool_calls', 아니면 None
    if getattr(last, "tool_calls", None):
    	return "tools"
	return END

 

 

5.그래프 구성

builder = StateGraph(State)
builder.add_node("call_model", call_model)
builder.add_node("tools", tool_node)

# connect nodes
builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", should_continue)
builder.add_edge("tools", "call_model")
graph = builder.compile()

 

 

6.실행

result = graph.invoke({"messages": [HumanMessage(content="LangGraph는 무엇인가요?")]})

for message in result["messages"]:
	message.pretty_print()

 

 

 

 

 

 

 

 

 

Agentic RAG

 

gpt기반 agent적 사고와 제어 흐름을 결합한 구조

• 스스로 판단하고, 도구를 선택하고, 여러번 검색하며 답을 능동적으로 구성하는 방식
• gpt가 검색, 판단, 재작성까지 담당하면서 정확하고 유연한 정보를 생성하는 강화된 rag 구조