S6 · 도구형 검색 에이전트 — create_agent 표준 패턴

Agentic RAG 실전 (14시간) · Day 2

S5에서 손으로 그린 agent ↔ tools 루프를 LangChain 1.x 표준 함수 create_agent 로 간결하게 만든다. 저수준 그래프와 고수준 함수의 선택 기준도 정리한다. S0 .venv/.env 로 실행.

목표

create_agent 로 도구형 에이전트를 만든다.
S5 손그래프와 비교해 무엇이 같고 무엇이 편해졌는지 안다.
여러 도구를 줬을 때 에이전트의 도구 선택을 확인한다.

1. 고수준 vs 저수준

저수준(S5): StateGraph 에 노드·엣지를 직접 구성. 모든 단계를 통제하지만 코드가 길다.
고수준(create_agent): 한 줄로 S5의 루프를 내부 생성. 도구 호출 판단→실행→재호출이 자동.

	저수준(StateGraph)	고수준(create_agent)
코드량	많음	적음
자유도	높음(커스텀 분기·루프)	표준 루프에 한정
적합	라우팅·자기교정 등 커스텀(S7·S8)	평범한 도구 호출

원칙: 표준 도구 호출이면 create_agent, 흐름을 특별히 손봐야 하면 저수준 StateGraph.

2. 버전 주의

인터넷 예제 다수는 구버전 langgraph.prebuilt.create_react_agent 를 쓴다. 1.x 표준은 from langchain.agents import create_agent 이며, ToolNode·tools_condition 같은 프리미티브는 여전히 langgraph.prebuilt 에 있다.

3. 실행 준비

S0 .venv 커널로 실행. .env 키 로드.

import os
from getpass import getpass

# S0 프로젝트 루트의 .env 에서 키 로드 (없으면 입력으로 폴백)
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    pass
for name in ["OPENAI_API_KEY", "PINECONE_API_KEY"]:
    if not os.environ.get(name):
        os.environ[name] = getpass(f"{name}: ")
print("키:", {k: bool(os.environ.get(k)) for k in ["OPENAI_API_KEY", "PINECONE_API_KEY"]})

키: {'OPENAI_API_KEY': True, 'PINECONE_API_KEY': True}

4. 인덱스 연결 & 도구

agentic-rag-lab 에 연결하고 S5와 동일한 검색 도구를 만든다.

import os
from pinecone import Pinecone
from langchain_pinecone import PineconeVectorStore, PineconeEmbeddings
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

embeddings = PineconeEmbeddings(model="multilingual-e5-large")   # 한국어 임베딩(1024차원)
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index = pc.Index("agentic-rag-lab")                              # S2/S3에서 색인한 인덱스
NAMESPACE = "chunk_500"                                          # S3에서 색인한 chunk_500 을 메인 코퍼스로 재사용한다. (S3 색인 셀을 먼저 실행할 것)
vector_store = PineconeVectorStore(index=index, embedding=embeddings, namespace=NAMESPACE)
llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0)            # 기본 LLM

# 데이터 유무 진단: 네임스페이스별 벡터 수 출력
try:
    stats = index.describe_index_stats()
    ns_map = getattr(stats, "namespaces", None) or (stats.get("namespaces") if isinstance(stats, dict) else {}) or {}
    ns_counts = {k: getattr(v, "vector_count", v.get("vector_count") if isinstance(v, dict) else None) for k, v in ns_map.items()}
    print("네임스페이스별 벡터 수:", ns_counts)
    if not ns_counts.get(NAMESPACE):
        print(f"⚠️ '{NAMESPACE}' 네임스페이스에 데이터가 없습니다. S3의 색인 셀을 먼저 실행하세요.")
except Exception as e:
    print("인덱스 통계 확인 생략:", e)
print("연결 완료")

네임스페이스별 벡터 수: {'chunk_1000': 18, 'chunk_500': 34, 'chunk_200': 89, 'chunk_1500': 12, 'demo': 3}
연결 완료

@tool
def search_docs(query: str) -> str:
    """사내 정책/기술/FAQ 문서에서 질문과 관련된 내용을 검색한다. 사내 규정·절차를 물을 때 사용."""
    hits = vector_store.similarity_search(query, k=3)          # 의미 검색 상위 3개
    if not hits:
        return "관련 문서를 찾지 못했습니다."
    return "\n\n".join(f"[{h.metadata['title']}] {h.page_content}" for h in hits)

5. create_agent 로 에이전트 생성

model: LLM · tools: 도구 목록 · system_prompt: 행동 지침(S5에서 SystemMessage로 붙이던 것) · checkpointer: 멀티턴 기억.

from langchain.agents import create_agent                    # ✅ 1.x 표준 위치
from langgraph.checkpoint.memory import InMemorySaver

SYSTEM = (
    "당신은 '누구나테크' 사내 문서 도우미입니다. "
    "사내 정책·기술·FAQ 질문이면 search_docs 도구로 근거를 찾아 답하고, "
    "인사말·일반 상식은 도구 없이 바로 답하세요. 답변 끝에 참고 문서 제목을 표기하세요."
)

agent = create_agent(                # 내부적으로 S5의 'agent↔tools 루프' 그래프를 생성
    model="gpt-4.1-mini",            # 사용할 LLM
    tools=[search_docs],             # 쥐여 줄 도구들
    system_prompt=SYSTEM,            # 행동 지침 (직접 SystemMessage 안 붙여도 됨)
    checkpointer=InMemorySaver(),    # 멀티턴 기억
)
print("에이전트 생성 완료 — 타입:", type(agent).__name__)   # CompiledStateGraph (우리가 컴파일한 것과 같은 종류)

에이전트 생성 완료 — 타입: CompiledStateGraph

6. 실행 — S5와 동일 동작

호출 방식은 손그래프와 같다: {"messages": [...]} + thread_id.

from langchain_core.messages import HumanMessage

def ask(question, thread):
    cfg = {"configurable": {"thread_id": thread}}
    result = agent.invoke({"messages": [HumanMessage(content=question)]}, cfg)
    msgs = result["messages"]
    # 체크포인터로 히스토리가 누적되므로, '이번 턴'(마지막 사용자 메시지 이후)만 확인한다.
    last_human = max(i for i, m in enumerate(msgs) if m.type == "human")
    used = any(getattr(m, "tool_calls", None) for m in msgs[last_human:])
    print("Q:", question)
    print("A:", msgs[-1].content)
    print("→ 검색 도구 사용:", used, "\n")

# 비교가 섞이지 않도록 질문마다 다른 thread 사용
ask("배포는 어떤 방식으로 진행되나요?", thread="q-search")     # 사내 규정 → 검색
ask("좋은 아침이에요!", thread="q-greeting")                  # 인사 → 직답

Q: 배포는 어떤 방식으로 진행되나요?
A: 배포는 QA 승인을 받고 릴리스 담당자가 배포 태그를 생성하면 엔지니어링 리드가 수동으로 최종 승인하는 2단계 게이트를 거칩니다. 기능 플래그로 코드 배포와 기능 노출을 분리하고, 위험 기능은 일부 고객에게 먼저 공개하는 점진적 출시(카나리 방식)를 적용합니다. 카나리 배포는 전체 트래픽 중 5%에 먼저 배포 후 오류율과 지연 지표를 관찰하며 25%, 50%, 100%로 단계 확대합니다. 오류율이 기준치를 넘으면 자동으로 배포가 중단되고 알림이 가며, 문제가 있을 시 롤백은 1분 이내 자동으로 진행됩니다. 긴급 핫픽스는 별도 승인 절차로 스테이징을 건너뛰고 배포 후 24시간 내 리뷰를 작성합니다. 정기 배포는 화·목 오전에만 진행하며 금요일과 공휴일 전날 오후는 배포 금지입니다.

참고: 배포 및 릴리스 프로세스
→ 검색 도구 사용: True 

Q: 좋은 아침이에요!
A: 좋은 아침입니다! 오늘도 좋은 하루 보내세요. 도움이 필요하시면 언제든 말씀해 주세요.
→ 검색 도구 사용: False

7. 여러 도구 — 에이전트의 선택

도구가 둘 이상이면 에이전트가 질문에 맞는 도구를 고른다(독스트링 기반).

@tool
def list_doc_titles() -> str:
    """사내 핸드북에 어떤 문서(주제)들이 있는지 제목 목록을 돌려준다. '무슨 문서가 있냐'고 물을 때 사용."""
    # 데모용으로 인덱스의 일부를 검색해 제목을 모은다(실무에선 메타데이터 조회로 대체 가능).
    seen, titles = set(), []
    for d in vector_store.similarity_search("정책 규정 절차 가이드", k=15):
        t = d.metadata.get("title")
        if t and t not in seen:
            seen.add(t); titles.append(t)
    return "사내 문서 목록: " + ", ".join(titles)

multi_agent = create_agent(
    model="gpt-4.1-mini",
    tools=[search_docs, list_doc_titles],     # 도구 2개
    system_prompt="사내 문서 도우미. 질문에 맞는 도구를 골라 사용하세요.",
)

for q in ["어떤 문서들이 있어요?", "연차는 며칠인가요?"]:
    r = multi_agent.invoke({"messages": [HumanMessage(content=q)]})
    # 어떤 도구가 호출됐는지 추출
    tools_used = [tc["name"] for m in r["messages"] for tc in (getattr(m, "tool_calls", None) or [])]
    print(f"Q: {q}\n  사용 도구: {tools_used}\n  A: {r['messages'][-1].content[:80]}...\n")

Q: 어떤 문서들이 있어요?
  사용 도구: ['list_doc_titles']
  A: 현재 사내 문서로는 배포 및 릴리스 프로세스, 데이터 거버넌스 및 개인정보 보호, 보안 및 계정 관리 규정, 장애 대응 및 온콜 운영, 신규 입...

Q: 연차는 며칠인가요?
  사용 도구: ['search_docs']
  A: 누구나테크의 연차 휴가는 입사 1년 차에 15일이 부여되며, 근속 2년마다 1일씩 가산되어 최대 25일까지 늘어납니다. 입사 첫해에는 매월 만근...

🔍 관찰 포인트

S5와 동작은 같고 코드는 짧다.
"어떤 문서가 있냐" → list_doc_titles, "연차 며칠" → search_docs. 도구 설명이 선택을 좌우한다.
커스텀 흐름(S7 라우팅, S8 교정)은 표준 루프에 안 맞아 저수준으로 돌아간다.

✏️ 미니 실습

search_docs 독스트링을 모호하게 바꿔 선택 정확도 변화를 본다.
같은 thread로 후속 질문해 멀티턴을 확인한다.
세 번째 도구(예: 날짜 반환)를 추가한다.

# 1) search_docs 독스트링을 모호하게 바꾸면 도구 선택 정확도가 흔들릴 수 있다.

@tool
def search_docs_vague(query: str) -> str:
    """질문에 도움이 되는 내용을 찾아준다."""
    hits = vector_store.similarity_search(query, k=3)
    if not hits:
        return "관련 문서를 찾지 못했습니다."
    return "\n\n".join(f"[{h.metadata['title']}] {h.page_content}" for h in hits)

multi_agent_vague = create_agent(
    model="gpt-4.1-mini",
    tools=[search_docs_vague, list_doc_titles],
    system_prompt="사내 문서 도우미. 질문에 맞는 도구를 골라 사용하세요.",

)

for q in ["어떤 문서들이 있어요?", "연차는 며칠인가요?"]:
    result = multi_agent_vague.invoke({"messages": [HumanMessage(content=q)]})
    tools_used = [tc["name"] for m in result["messages"] for tc in (getattr(m, "tool_calls", None) or [])]
    print(f"Q: {q}\n  사용 도구: {tools_used}\n  A: {result['messages'][-1].content[:80]}...\n")

Q: 어떤 문서들이 있어요?
  사용 도구: ['list_doc_titles']
  A: 사내 문서로는 배포 및 릴리스 프로세스, 데이터 거버넌스 및 개인정보 보호, 보안 및 계정 관리 규정, 장애 대응 및 온콜 운영, 신규 입사자 ...

Q: 연차는 며칠인가요?
  사용 도구: ['list_doc_titles', 'search_docs_vague']
  A: 연차는 입사 1년 차에 15일이 부여되며, 근속 2년마다 1일씩 가산되어 최대 25일까지 늘어납니다. 입사 첫해에는 매월 만근 시 1일씩 비례하...

실행 결과 해설

어떤 문서들이 있어요?에서는 list_doc_titles만 호출된다. 질문이 '목록' 요청이라서 제목 나열 도구 설명과 더 잘 맞기 때문이다.
연차는 며칠인가요?에서는 search_docs_vague만 깔끔하게 선택되지 않고 list_doc_titles가 함께 섞여 호출될 수 있다. search_docs_vague의 설명이 질문에 도움이 되는 내용을 찾아준다처럼 너무 넓어서, 모델이 어떤 상황에 이 도구를 써야 하는지 경계를 분명히 잡지 못했기 때문이다.
핵심은 도구 설명이 단순한 주석이 아니라 선택 기준이라는 점이다. 도구 수가 늘어날수록 독스트링은 언제 쓰는가, 무엇을 반환하는가를 구체적으로 적어야 선택 정확도가 올라간다.

# 2) 같은 thread_id(대화 스레드 식별자)로 후속 질문하면 앞 맥락이 유지된다.

multi_agent_memory = create_agent(
    model="gpt-4.1-mini",
    tools=[search_docs, list_doc_titles],
    system_prompt="사내 문서 도우미. 질문에 맞는 도구를 골라 사용하세요.",
    checkpointer=InMemorySaver(),
)

def ask_multi(question, thread):
    cfg = {"configurable": {"thread_id": thread}}
    result = multi_agent_memory.invoke({"messages": [HumanMessage(content=question)]}, cfg)
    print("Q:", question)
    print("A:", result["messages"][-1].content)
    print()

ask_multi("배포는 어떤 방식으로 진행되나요?", thread="multi")
ask_multi("그 절차에서 승인은 누가 하나요?", thread="multi")

Q: 배포는 어떤 방식으로 진행되나요?
A: 배포는 QA 승인 이후 릴리스 담당자가 배포 태그를 생성하고, 엔지니어링 리드가 수동으로 승인하는 2단계 게이트를 거쳐 진행됩니다. 기능 플래그를 활용해 코드 배포와 기능 노출을 분리하며, 위험한 기능은 부분 고객에게 먼저 공개하는 점진적 출시를 적용합니다.

배포는 카나리 방식으로 전체 트래픽의 5%에 먼저 적용하고, 오류율과 지연 지표를 10분간 관찰 후 25%, 50%, 100% 순으로 확대합니다. 오류율이 기준치(0.5%)를 초과하면 자동으로 확대를 중단하고 알림이 갑니다. 주요 지표는 실시간 대시보드로 모니터링하며, 배포 시 담당자가 대기합니다.

문제가 발견되면 직전 배포 태그로 자동 롤백하며, 긴급 핫픽스는 별도 승인 후 스테이징을 건너뛸 수 있으나 사후 리뷰가 필요합니다. 정기 배포는 화·목요일 오전에만 진행하며, 금요일 및 공휴일 전날 오후에는 배포가 금지되어 있습니다.

Q: 그 절차에서 승인은 누가 하나요?
A: 배포 절차에서 승인은 릴리스 담당자가 배포 태그를 생성한 후, 엔지니어링 리드가 수동으로 최종 승인을 합니다. 이 2단계 게이트를 통해 배포가 진행됩니다.

실행 결과 해설

두 질문이 모두 같은 thread_id를 사용하므로, 두 번째 질문은 첫 번째 질문의 답을 바탕으로 해석된다. 그래서 그 절차가 무엇인지 다시 설명하지 않아도 배포 절차를 가리키는 것으로 이해한다.
이 동작은 checkpointer=InMemorySaver()가 대화 상태를 스레드별로 저장하기 때문에 가능하다. 즉, 멀티턴은 모델이 우연히 기억한 것이 아니라 체크포인터가 이전 메시지를 다시 공급해 주는 구조다.
실무에서는 사용자별 세션 아이디를 thread_id로 쓰면 된다. 반대로 새 대화처럼 다뤄야 하는 요청에는 다른 thread_id를 써야 이전 문맥이 섞이지 않는다.

# 3) 세 번째 도구를 추가하면 질문 유형에 따라 그 도구도 선택할 수 있다.
from datetime import date

@tool
def today_date() -> str:
    """오늘 날짜를 YYYY-MM-DD 형식으로 반환한다. 날짜를 묻는 질문에 사용."""
    return date.today().isoformat()

multi_agent_3tools = create_agent(
    model="gpt-4.1-mini",
    tools=[search_docs, list_doc_titles, today_date],
    system_prompt="사내 문서 도우미. 질문에 맞는 도구를 골라 사용하세요.",
)

for q in ["오늘 날짜가 뭐예요?", "어떤 문서들이 있어요?", "배포는 어떤 방식으로 진행되나요?"]:
    result = multi_agent_3tools.invoke({"messages": [HumanMessage(content=q)]})
    tools_used = [tc["name"] for m in result["messages"] for tc in (getattr(m, "tool_calls", None) or [])]
    print(f"Q: {q}\n  사용 도구: {tools_used}\n  A: {result['messages'][-1].content[:80]}...\n")

Q: 오늘 날짜가 뭐예요?
  사용 도구: ['today_date']
  A: 오늘 날짜는 2026년 6월 15일입니다....

Q: 어떤 문서들이 있어요?
  사용 도구: ['list_doc_titles']
  A: 사내 문서로는 다음과 같은 것들이 있습니다:
- 배포 및 릴리스 프로세스
- 데이터 거버넌스 및 개인정보 보호
- 보안 및 계정 관리 규정
- ...

Q: 배포는 어떤 방식으로 진행되나요?
  사용 도구: ['search_docs']
  A: 배포는 QA 승인 후 릴리스 담당자가 배포 태그를 생성하고 엔지니어링 리드가 수동으로 승인하는 2단계 게이트를 거칩니다. 기능 플래그를 활용해 ...

실행 결과 해설

오늘 날짜가 뭐예요?에는 today_date, 어떤 문서들이 있어요?에는 list_doc_titles, 배포는 어떤 방식으로 진행되나요?에는 search_docs가 선택된다. 즉, 같은 에이전트가 질문 유형에 따라 서로 다른 도구를 고른다.
여기서 중요한 기준은 도구의 이름보다 설명이다. today_date는 날짜 반환, list_doc_titles는 문서 제목 목록, search_docs는 문서 내용 검색이라는 역할 경계가 분명해서 선택이 안정적이다.
이 패턴이 표준 도구형 에이전트의 핵심이다. 흐름을 직접 분기하지 않아도, 도구 설명만 잘 설계하면 상당수 질의는 create_agent의 기본 루프만으로 처리할 수 있다.

정리

create_agent 로 도구형 에이전트를 간결히 만들고 S5와 비교했다.
고수준은 빠르고 저수준은 자유롭다 — 상황에 맞게 고른다.
다음(S7): 표준 루프로 부족한 라우팅·쿼리 재작성(Adaptive RAG)을 저수준으로 구현한다.

저작자표시 비영리 변경금지 (새창열림)

'AI System > Agentic RAG: Agent 시스템을 위한 Vector RAG 설계' 카테고리의 다른 글

d02 - 9. CRAG & Self-RAG (0)	2026.06.16
d02 - 8. Adaptive RAG (0)	2026.06.16
d01 - 6. AgenticRAG 실습 (0)	2026.06.15
d01 - 5. LangGraph (0)	2026.06.15
d01 - 4. RAG 파이프라인 (0)	2026.06.15

Toddler_AD 님의 블로그

d02 - 7. create_agent

S6 · 도구형 검색 에이전트 — create_agent 표준 패턴

목표

1. 고수준 vs 저수준

2. 버전 주의

3. 실행 준비

4. 인덱스 연결 & 도구

5. create_agent 로 에이전트 생성

6. 실행 — S5와 동일 동작

7. 여러 도구 — 에이전트의 선택

🔍 관찰 포인트

✏️ 미니 실습

실행 결과 해설

실행 결과 해설

실행 결과 해설

정리

'AI System > Agentic RAG: Agent 시스템을 위한 Vector RAG 설계' 카테고리의 다른 글

티스토리툴바

d02 - 7. create_agent

S6 · 도구형 검색 에이전트 — create_agent 표준 패턴

목표

1. 고수준 vs 저수준

2. 버전 주의

3. 실행 준비

4. 인덱스 연결 & 도구

5. create_agent 로 에이전트 생성

6. 실행 — S5와 동일 동작

7. 여러 도구 — 에이전트의 선택

🔍 관찰 포인트

✏️ 미니 실습

실행 결과 해설

실행 결과 해설

실행 결과 해설

정리

'AI System > Agentic RAG: Agent 시스템을 위한 Vector RAG 설계' 카테고리의 다른 글

관련글

티스토리툴바