[LangGraph] A2A 멀티에이전트 구현하기 — 오케스트레이터, RAG, 멀티쿼리

1. 패키지 구조

1) 전체 구조

Morning-Agent-Hub/
├── ai_llm/
│   ├── orchestrator_agent/
│   │   ├── agent.py          ← 인텐트 분류, 플랜 실행
│   │   └── agent_server.py   ← A2A HTTP 서버
│   ├── internal_rag_agent/
│   │   ├── agent.py          ← LangGraph 그래프
│   │   ├── document_parser.py    ← 문서 파싱
│   │   ├── indexing_service.py   ← 청크 분할, 임베딩, DB 저장
│   │   ├── rag_search_service.py ← 멀티쿼리, 벡터 검색, 답변 생성
│   │   └── agent_server.py
│   ├── web_research_agent/
│   ├── file_management_agent/
│   └── report_writing_agent/
├── backend/
│   └── api/
│       └── routers/
│           └── chat.py       ← FastAPI 엔드포인트
├── frontend/
│   └── src/
│       └── components/
│           └── ChatView.vue
└── start_agents.py           ← 전체 에이전트 실행

 

2) 구조 설명

• 에이전트마다 agent.py(핵심 로직)와 agent_server.py(A2A HTTP 서버)를 분리했다
• RAG 에이전트는 역할별로 3개 모듈로 추가 분리했다
• backend는 Vue.js 프론트엔드와 에이전트 사이의 중간 레이어 역할을 한다

 

2. 에이전트 실행

1) start_agents.py

• 모든 에이전트는 start_agents.py로 한 번에 실행한다. 원격 에이전트가 먼저 초기화된 뒤 오케스트레이터가 마지막으로 실행된다.

import subprocess, sys, time, os
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parent
AI_LLM_ROOT  = PROJECT_ROOT / "ai_llm"

AGENTS = {
    "web":          {"port": 10011, "dir": AI_LLM_ROOT / "web_research_agent"},
    "rag":          {"port": 10012, "dir": AI_LLM_ROOT / "internal_rag_agent"},
    "file":         {"port": 10013, "dir": AI_LLM_ROOT / "file_management_agent"},
    "report":       {"port": 10014, "dir": AI_LLM_ROOT / "report_writing_agent"},
    "orchestrator": {"port": 10010, "dir": AI_LLM_ROOT / "orchestrator_agent"},
}

def start_agent(key: str) -> subprocess.Popen:
    agent = AGENTS[key]
    env   = {**os.environ, "PYTHONPATH": str(PROJECT_ROOT)}
    return subprocess.Popen(
        [sys.executable, str(agent["dir"] / "agent_server.py")],
        cwd=str(agent["dir"]),
        env=env,
    )

# 원격 에이전트 먼저 실행
for key in ["web", "rag", "file", "report"]:
    start_agent(key)
    time.sleep(2)

# 5초 대기 후 오케스트레이터 실행
time.sleep(5)
start_agent("orchestrator")

 

2) 코드 설명

• PYTHONPATH를 PROJECT_ROOT로 설정하여 모든 에이전트에서 공통 모듈을 import할 수 있게 한다
• 원격 에이전트를 먼저 실행하고 5초 대기하는 이유는 오케스트레이터가 실행될 때 원격 에이전트의 Agent Card를 읽어야 하기 때문이다
• 에이전트 사이에 time.sleep(2)를 두는 이유는 포트 바인딩 충돌을 방지하기 위해서다

 

3. 오케스트레이터 구현

1) 인텐트 분류

• gpt-4o-mini의 Structured Output으로 IntentPlan 객체를 생성한다.

from enum import Enum
from pydantic import BaseModel
from openai import AsyncOpenAI

client = AsyncOpenAI()

class Intent(str, Enum):
    INTERNAL_SEARCH = "INTERNAL_SEARCH"
    WEB_SEARCH      = "WEB_SEARCH"
    FILE_OPERATION  = "FILE_OPERATION"
    HYBRID          = "HYBRID"
    DIRECT          = "DIRECT"

class PlanStep(BaseModel):
    agent:       str
    instruction: str
    depends_on:  list[int] = []

class IntentPlan(BaseModel):
    intent: Intent
    steps:  list[PlanStep]

async def classify_intent(message: str) -> IntentPlan:
    response = await client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": ORCHESTRATOR_SYSTEM_PROMPT},
            {"role": "user",   "content": message},
        ],
        response_format=IntentPlan,
    )
    return response.choices[0].message.parsed

 

2) 플랜 실행

• depends_on이 없는 단계는 asyncio.gather로 병렬 실행하고, 의존 관계가 있는 단계는 이전 결과를 받아 순차 실행한다.

import asyncio

async def execute_plan(plan: IntentPlan) -> dict:
    results = {}

    independent = [s for s in plan.steps if not s.depends_on]
    dependent   = [s for s in plan.steps if s.depends_on]

    # 독립 단계 병렬 실행
    parallel_results = await asyncio.gather(*[
        call_agent(step.agent, step.instruction) for step in independent
    ])
    for step, result in zip(independent, parallel_results):
        results[step.agent] = result

    # 의존 단계 순차 실행
    for step in dependent:
        prior = "\n".join(results[plan.steps[i].agent] for i in step.depends_on)
        results[step.agent] = await call_agent(
            step.agent,
            f"{step.instruction}\n\n이전 결과:\n{prior}"
        )

    return results

3) RAG → 웹 자동 폴백

RAG 결과가 없을 때 반환되는 문구를 패턴 매칭으로 감지하여 웹 검색으로 자동 전환한다.

def _is_rag_no_result(text: str) -> bool:
    patterns = [
        "현재 인덱싱된 문서에서 관련 내용을 찾지 못했습니다",
        "관련 문서를 찾을 수 없습니다",
        "검색 결과가 없습니다",
    ]
    return any(p in text for p in patterns)

async def handle_internal_search(message: str) -> str:
    rag_result = await call_agent("rag", message)

    if _is_rag_no_result(rag_result):
        await stream_status("📄 사내 문서에서 찾지 못했습니다. 웹에서 검색합니다...")
        web_result = await call_agent("web", message)

        if web_result.get("max_score", 0) >= 0.5:
            return f"🌐 [웹 검색 기반]\n{web_result['content']}"
        return "관련 정보를 찾을 수 없습니다."

    return f"📄 [사내 문서 기반]\n{rag_result}"

4) 환각 방지

• 초기 구현에서 모든 에이전트 결과가 generate_final_response()를 통해 gpt-4o로 재합성됐다. RAG가 문서 기반 답변을 정확히 찾아도 gpt-4o가 학습 데이터로 재합성하는 과정에서 문서에 없는 내용이 섞이는 환각이 발생했다.
• RAG, 웹, 보고서 결과는 에이전트 결과를 직접 반환하고, 여러 에이전트를 통합해야 하는 HYBRID 작업에서만 gpt-4o 재합성을 사용한다.

async def process_result(intent: Intent, agent_results: dict) -> str:
    if intent == Intent.INTERNAL_SEARCH:
        return agent_results.get("rag", "")

    if intent == Intent.WEB_SEARCH:
        return agent_results.get("web", "")

    if intent == Intent.FILE_OPERATION:
        return agent_results.get("report", "")

    # HYBRID만 gpt-4o로 통합 요약
    if intent == Intent.HYBRID:
        return await generate_final_response(agent_results)

    return await direct_response(agent_results.get("message", ""))

 

4. RAG 에이전트 — LangGraph 그래프

1) State 정의

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class RAGState(TypedDict):
    query:          str
    is_contextual:  bool      # 맥락 의존 질문 여부
    intent:         str       # search | index
    chunks:         list[dict]
    answer:         str
    sources:        list[dict]

 

2) 그래프 빌드

builder = StateGraph(RAGState)

builder.add_node("contextualize_question", contextualize_question_node)
builder.add_node("intent_router",          intent_router_node)
builder.add_node("vector_search",          vector_search_node)
builder.add_node("generate",               generate_node)
builder.add_node("index_document_node",    index_document_node)

builder.add_edge(START, "contextualize_question")
builder.add_edge("contextualize_question", "intent_router")

builder.add_conditional_edges("intent_router", route_intent, {
    "search": "vector_search",
    "index":  "index_document_node",
})

builder.add_edge("vector_search", "generate")
builder.add_edge("generate",      END)
builder.add_edge("index_document_node", END)

graph = builder.compile()

 

3) contextualize_question 노드

• START 직후 실행되는 전처리 노드로, 포맷 아티팩트 제거와 맥락 의존 질문 감지를 담당한다.

import re

CONTEXT_MARKERS = ["그것", "그게", "위에서", "방금", "아까", "해당 내용", "거기서"]

def contextualize_question_node(state: RAGState) -> RAGState:
    query = state["query"]

    # [이전 대화] / [현재 질문] 포맷 아티팩트 제거
    query = re.sub(r"\[이전 대화\].*?\[현재 질문\]", "", query, flags=re.DOTALL).strip()

    # 맥락 의존 질문 감지
    is_contextual = any(marker in query for marker in CONTEXT_MARKERS)

    return {**state, "query": query, "is_contextual": is_contextual}

• [이전 대화], [현재 질문] 같은 포맷 아티팩트가 벡터 검색 쿼리에 섞이면 임베딩 품질이 저하된다
• "그것", "위에서" 같은 지시어가 포함된 질문은 이전 대화 없이 검색하면 실패할 가능성이 높다
• is_contextual이 True이고 검색 결과가 없으면 재질문 유도 메시지를 반환한다

 

5. RAG 에이전트 — 멀티쿼리 검색

1) 전체 코드

async def rewrite_query(query: str) -> str:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role":    "system",
                "content": "벡터 검색에 최적화된 형태로 쿼리를 재작성하라. 조사를 제거하고 핵심 명사 중심으로 확장하라.",
            },
            {"role": "user", "content": query},
        ],
    )
    return response.choices[0].message.content.strip()


async def generate_multi_queries(query: str) -> list[str]:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role":    "system",
                "content": "아래 쿼리의 의미는 유지하되 다른 표현으로 변형 쿼리 3개를 생성하라. 각 줄에 하나씩 출력하라.",
            },
            {"role": "user", "content": query},
        ],
    )
    return response.choices[0].message.content.strip().split("\n")[:3]


async def multi_query_search(query: str) -> list[dict]:
    # 쿼리 재작성 + 변형 3개 생성 → 총 4개
    rewritten   = await rewrite_query(query)
    variants    = await generate_multi_queries(rewritten)
    all_queries = [rewritten] + variants

    # 4개 쿼리 병렬 임베딩
    embeddings = await asyncio.gather(*[
        client.embeddings.create(model="text-embedding-3-small", input=q)
        for q in all_queries
    ])
    vectors = [r.data[0].embedding for r in embeddings]

    # Supabase RPC 병렬 호출
    results = await asyncio.gather(*[
        asyncio.to_thread(
            lambda v=vec: supabase.rpc("match_documents", {
                "query_embedding": v,
                "match_threshold": 0.3,
                "match_count":     10,
            }).execute()
        )
        for vec in vectors
    ])

    # storage_ref:chunk_index 키로 중복 제거 (높은 유사도 우선)
    seen: dict[str, dict] = {}
    for result in results:
        for chunk in result.data:
            key = f"{chunk['storage_ref']}:{chunk['chunk_index']}"
            if key not in seen or chunk["similarity"] > seen[key]["similarity"]:
                seen[key] = chunk

    # 유사도 내림차순 상위 10개 반환
    return sorted(seen.values(), key=lambda x: x["similarity"], reverse=True)[:10]

 

2) 코드 설명

(1) rewrite_query

• 불필요한 조사("~은", "~가")를 제거하고 핵심 명사 중심으로 쿼리를 확장한다
• "3단계"처럼 특정 표현 대신 "1단계 2단계 3단계" 전체를 포함시켜 문서 표현과 매칭 범위를 넓힌다

(2) generate_multi_queries

• 같은 의미지만 문서에서 쓸 수 있는 다양한 표현으로 변형 쿼리 3개를 생성한다
• 4개 쿼리로 검색하면 다양한 표현 방식으로 작성된 문서 청크를 더 많이 찾을 수 있다

(3) 중복 제거

• 4개 쿼리 결과는 최대 40개의 청크를 반환한다
• 같은 청크가 여러 쿼리에서 반환되면 가장 높은 유사도 점수만 유지하고 나머지는 버린다
• 중복이 많을수록 해당 청크가 여러 표현으로 질문과 관련이 높다는 신호다

 

6. 문서 인덱싱

1) 전체 코드

from langchain_text_splitters import RecursiveCharacterTextSplitter
from enum import Enum
from dataclasses import dataclass

class IndexJobStatus(str, Enum):
    success = "success"
    skipped = "skipped"
    error   = "error"

@dataclass
class IndexJob:
    status:      IndexJobStatus
    chunk_count: int  = 0
    error:       str  = ""

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=100,
)

async def index_document(file_path: str, storage_ref: str) -> IndexJob:
    # 중복 인덱싱 방지
    if await _is_already_indexed(storage_ref):
        return IndexJob(status=IndexJobStatus.skipped)

    try:
        # 텍스트 파싱 — 각 페이지 앞에 [페이지 N] 마커 삽입
        text   = parse_document(file_path)
        chunks = splitter.split_text(text)

        # 임베딩 생성
        response = await client.embeddings.create(
            model="text-embedding-3-small",
            input=chunks,
        )
        embeddings = [r.embedding for r in response.data]

        # Supabase 배치 INSERT
        rows = [
            {
                "content":     chunk,
                "embedding":   emb,
                "storage_ref": storage_ref,
                "chunk_index": i,
            }
            for i, (chunk, emb) in enumerate(zip(chunks, embeddings))
        ]
        await asyncio.to_thread(
            lambda: supabase.table("documents").insert(rows).execute()
        )

        return IndexJob(status=IndexJobStatus.success, chunk_count=len(chunks))

    except Exception as e:
        return IndexJob(status=IndexJobStatus.error, error=str(e))

 

2) 코드 설명

• chunk_size를 500에서 800으로 키운 이유는 500자로 분할하면 등급별 항목처럼 연속된 정보가 청크 경계에서 잘려 검색 품질이 낮아졌기 때문이다
• 초기 구현에서는 청크마다 개별 INSERT를 수행하여 100개 청크 기준 100번의 DB 왕복이 발생했다. 배치 INSERT로 변경하여 1번의 왕복으로 처리한다
• Supabase Python SDK의 .execute()는 동기 함수다. FastAPI 비동기 컨텍스트에서 직접 호출하면 이벤트 루프가 블로킹되므로 asyncio.to_thread()로 별도 스레드에서 실행한다

 

7. 출처 표시

1) 페이지 번호 추출

청크 content 내 [페이지 N] 마커를 정규식으로 파싱하여 페이지 번호를 추출한다. 동일 파일의 여러 청크는 페이지 번호를 집계하여 하나로 묶는다.

import re

def _extract_pages(content: str) -> list[int]:
    return [int(n) for n in re.findall(r"\[페이지 (\d+)\]", content)]

def _format_page_range(pages: list[int]) -> str:
    if not pages:
        return ""
    if len(pages) == 1:
        return f"p.{pages[0]}"
    return f"p.{pages[0]}–{pages[-1]}"

def build_source_section(chunks: list[dict]) -> str:
    file_pages: dict[str, set[int]] = {}
    file_urls:  dict[str, str]      = {}

    for chunk in chunks:
        ref   = chunk["storage_ref"]
        pages = _extract_pages(chunk["content"])
        file_pages.setdefault(ref, set()).update(pages)
        file_urls[ref] = chunk.get("file_url", "")

    lines = ["**출처:**"]
    for ref, pages in file_pages.items():
        page_str = _format_page_range(sorted(pages))
        lines.append(f"- **{ref}** ({page_str})")
        if file_urls[ref]:
            lines.append(f"  파일 위치: {file_urls[ref]}")

    return "\n".join(lines)

 

2) 출력 예시

**출처:**
- **공공데이터 가이드라인.pdf** (p.13–15)
  파일 위치: https://drive.google.com/file/d/xxx/view
- **AI 활용 전략.pdf** (p.4)
  파일 위치: https://drive.google.com/file/d/yyy/view

 

 

8. Web Research 에이전트

1) 구현

• LangGraph의 create_react_agent를 사용하며 Tavily 도구는 MCP 방식으로 연결한다.

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

async def build_web_agent():
    mcp_client = MultiServerMCPClient({
        "tavily": {
            "command":   "npx",
            "args":      ["-y", "tavily-mcp@0.1.4"],
            "env":       {"TAVILY_API_KEY": os.getenv("TAVILY_API_KEY")},
            "transport": "stdio",
        }
    })
    tools = await mcp_client.get_tools()
    model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    return create_react_agent(model, tools)

 

2) 관련성 점수 기반 필터링

def filter_by_relevance(results: list[dict]) -> list[dict]:
    if not results:
        return []

    max_score = max(r.get("score", 0) for r in results)

    # 최고 score가 0.5 미만이면 관련 정보 없음
    if max_score < 0.5:
        return []

    return [r for r in results if r.get("score", 0) >= 0.5]

• Tavily는 검색 결과마다 score 필드(0~1)를 반환한다
• score가 낮을수록 검색 결과가 질문과 무관하다는 의미다
• 존재하지 않는 개념을 질문하면 Tavily가 관련 없는 문서를 억지로 반환하는데 이때 모든 score가 0.5 미만으로 나타난다