계획하고, 기억하고, 채점하기

 1[Day7] 공통 checkpoint
 2
 37일차에 실행하는 완성 코드가 모두 들어 있습니다.
 4day7_practice 폴더에 풀어 두고 씁니다.
 5
 6담긴 것
 7  plan_and_execute.py    1번째 세션 Plan-and-Execute 완성본
 8  long_term_memory.py    2번째 세션 장기 기억(Store) 완성본
 9  local_tracer.py        3번째 세션 실행 기록 콜백 핸들러
10  trace_run.py           3번째 세션 6일차 그래프에 추적 붙이기
11  llm_judge.py           4번째 세션 LLM 심사
12  run_eval.py            4번째 세션 누적 평가셋 실행
13  integration_check.py   6번째 세션 통합 점검
14  final_scenario.py      6일차 Supervisor 종합 시나리오 (RAG + MCP)
15  mcp_server.py          4일차 MCP 서버
16  make_data.py           임직원·자산·프로젝트 JSON 3종 생성기
17  guards_input.py        5일차 입력 가드레일 (run_eval, integration_check 가 import)
18  guards_refusal.py      5일차 고정 거절 문구 (위와 같은 이유)
19  eval_set.json          2~6일차 누적 평가셋 14문항 (전날 미션을 못 따라온 경우에만 사용)
20
21시작 전에 할 것
22  1) python make_data.py 로 JSON 3종 생성
23  2) 2일차 chroma_db 폴더를 day6_practice 에서 복사해 옵니다 (final_scenario 가 씁니다)
24  3) 레포 루트의 가상환경을 그대로 활성화합니다
25       ..\.venv\Scripts\activate
26  4) .env 는 레포 루트의 것을 씁니다
27
28주의
29  guards_input.py 와 guards_refusal.py 는 5일차에 직접 만든 본인 파일이 있으면
30  그것으로 덮어써서 쓰세요. 여기 것은 5일차를 못 따라온 경우의 대체본입니다.

 1# make_data.py - Day4 백데이터 3종을 만드는 스크립트 (교육용 더미 데이터)
 2"""실행하면 같은 폴더에 employees.json, assets.json, projects.json을 만듭니다.
 3
 4실행: python make_data.py
 5
 65, 6, 7일차가 이 데이터를 그대로 이어 쓰므로 값은 바꾸지 마세요.
 7"""
 8import json
 9import os
10
11BASE = os.path.dirname(os.path.abspath(__file__))
12
13EMPLOYEES = [
14    {"id": 1, "name": "김하늘", "team": "클라우드운영팀", "role": "팀장", "email": "haneul.kim@samsungsds.example.com"},
15    {"id": 2, "name": "박도윤", "team": "물류플랫폼팀", "role": "백엔드", "email": "doyun.park@samsungsds.example.com"},
16    {"id": 3, "name": "이서준", "team": "클라우드운영팀", "role": "인프라 엔지니어", "email": "seojun.lee@samsungsds.example.com"},
17    {"id": 4, "name": "최민준", "team": "물류플랫폼팀", "role": "프론트엔드", "email": "minjun.choi@samsungsds.example.com"},
18    {"id": 5, "name": "정하은", "team": "클라우드운영팀", "role": "데이터 분석가", "email": "haeun.jung@samsungsds.example.com"},
19]
20
21# 노트북 5대, 모니터 2대, GPU 서버 1대로 총 8건입니다. GPU 서버는 6일차 MCP 통합 실습이 조회하므로 빼지 마세요.
22# 5-2의 총액 계산(노트북 5대 x 200만원 = 1,000만원)이 이 구성에 맞춰져 있으니 건수를 바꾸지 마세요.
23ASSETS = [
24    {"asset_id": "A-1001", "type": "노트북", "model": "갤럭시북4 프로", "owner": "김하늘", "status": "사용중"},
25    {"asset_id": "A-1002", "type": "노트북", "model": "갤럭시북4 프로", "owner": "박도윤", "status": "사용중"},
26    {"asset_id": "A-1003", "type": "노트북", "model": "갤럭시북4", "owner": "이서준", "status": "사용중"},
27    {"asset_id": "A-1004", "type": "노트북", "model": "갤럭시북4", "owner": None, "status": "재고"},
28    {"asset_id": "A-1005", "type": "노트북", "model": "갤럭시북4 프로", "owner": "최민준", "status": "사용중"},
29    {"asset_id": "A-1006", "type": "모니터", "model": "뷰피니티 S8", "owner": "김하늘", "status": "사용중"},
30    {"asset_id": "A-1007", "type": "모니터", "model": "뷰피니티 S8", "owner": "정하은", "status": "사용중"},
31    {"asset_id": "A-1008", "type": "GPU 서버", "model": "NVIDIA A100 80GB", "owner": "이서준", "status": "사용중"},
32]
33
34PROJECTS = [
35    {"project_id": "P-2601", "name": "차세대 물류 플랫폼 구축", "team": "물류플랫폼팀", "status": "진행중", "member_count": 12},
36    {"project_id": "P-2602", "name": "클라우드 비용 최적화", "team": "클라우드운영팀", "status": "진행중", "member_count": 7},
37    {"project_id": "P-2603", "name": "물류 관제 대시보드 고도화", "team": "물류플랫폼팀", "status": "진행중", "member_count": 5},
38    {"project_id": "P-2604", "name": "사내 AI 어시스턴트 도입", "team": "클라우드운영팀", "status": "진행중", "member_count": 4},
39]
40
41
42def write_json(filename: str, rows: list) -> str:
43    """한글이 깨지지 않게 UTF-8과 ensure_ascii=False로 씁니다."""
44    path = os.path.join(BASE, filename)
45    with open(path, "w", encoding="utf-8") as f:
46        json.dump(rows, f, ensure_ascii=False, indent=2)
47    return path
48
49
50if __name__ == "__main__":
51    for filename, rows in [("employees.json", EMPLOYEES),
52                           ("assets.json", ASSETS),
53                           ("projects.json", PROJECTS)]:
54        write_json(filename, rows)
55        print(f"생성: {filename} ({len(rows)}건)")
56
57    laptops = [a for a in ASSETS if a["type"] == "노트북"]
58    total = len(laptops) * 2000000
59    print(f"확인: 자산 {len(ASSETS)}건 중 노트북 {len(laptops)}대, 한 대당 200만원이면 총액 {total:,}원")

 1# mcp_server.py - 삼성SDS 교육용 더미 사내 데이터 MCP 서버 (fastmcp)
 2"""노출하는 도구 4개:
 3  - search_employee(name)      임직원 개인 조회
 4  - list_team_members(team)    팀별 임직원 목록
 5  - get_asset(owner)           자산 조회
 6  - list_projects(team)        프로젝트 목록
 7
 8실행: python mcp_server.py
 9(stdio 모드로 대기합니다. 단독으로 쓰는 파일이 아니라
10 클라이언트가 자식 프로세스로 띄워서 사용합니다)
11"""
12import json
13import os
14from fastmcp import FastMCP
15
16mcp = FastMCP("sds-company-data")
17
18# make_data.py가 만든 JSON 3개를 코드와 같은 폴더에서 로드합니다 (교육용 더미 데이터)
19BASE = os.path.dirname(os.path.abspath(__file__))
20EMPLOYEES = json.load(open(os.path.join(BASE, "employees.json"), encoding="utf-8"))
21ASSETS = json.load(open(os.path.join(BASE, "assets.json"), encoding="utf-8"))
22PROJECTS = json.load(open(os.path.join(BASE, "projects.json"), encoding="utf-8"))
23
24
25@mcp.tool()
26def search_employee(name: str) -> str:
27    """삼성SDS 임직원 디렉터리에서 이름으로 직원 정보를 조회한다.
28
29    직원의 소속 팀, 직무, 이메일이 필요할 때 사용한다.
30    팀 전체 명단이나 인원수에는 사용하지 않는다 (list_team_members 사용).
31    자산이나 프로젝트 질문에도 사용하지 않는다 (get_asset, list_projects 사용).
32
33    Args:
34        name: 직원의 정확한 한국어 이름
35    """
36    for emp in EMPLOYEES:
37        if emp["name"] == name:
38            return json.dumps(emp, ensure_ascii=False)
39    names = ", ".join(e["name"] for e in EMPLOYEES)
40    return f"'{name}' 직원을 찾을 수 없습니다. 등록된 직원: {names}"
41
42
43@mcp.tool()
44def list_team_members(team: str) -> str:
45    """팀 이름으로 소속 임직원 목록을 조회한다.
46
47    특정 팀의 인원수나 명단이 필요할 때 사용한다.
48    개별 직원의 상세 정보는 search_employee를 사용한다.
49
50    Args:
51        team: 팀 이름 (예: '클라우드운영팀', '물류플랫폼팀')
52    """
53    rows = [e for e in EMPLOYEES if e["team"] == team]
54    if not rows:
55        teams = sorted({e["team"] for e in EMPLOYEES})
56        return f"'{team}' 팀을 찾을 수 없습니다. 등록된 팀: {teams}"
57    return json.dumps({"count": len(rows), "members": rows}, ensure_ascii=False, indent=2)
58
59
60@mcp.tool()
61def get_asset(owner: str = "") -> str:
62    """사내 자산(노트북, 모니터 등)을 조회한다.
63
64    Args:
65        owner: 사용자 이름으로 필터. 빈 문자열이면 전체 자산 반환
66    """
67    rows = [a for a in ASSETS if not owner or a["owner"] == owner]
68    if not rows:
69        return f"'{owner}' 사용자의 자산이 없습니다."
70    return json.dumps({"count": len(rows), "assets": rows}, ensure_ascii=False, indent=2)
71
72
73@mcp.tool()
74def list_projects(team: str = "") -> str:
75    """진행 중인 프로젝트 목록을 조회한다.
76
77    Args:
78        team: 팀 이름으로 필터 (예: '물류플랫폼팀'). 빈 문자열이면 전체
79    """
80    rows = [p for p in PROJECTS if not team or p["team"] == team]
81    if not rows:
82        return f"'{team}' 팀의 프로젝트가 없습니다."
83    return json.dumps(rows, ensure_ascii=False, indent=2)
84
85
86if __name__ == "__main__":
87    mcp.run()   # stdio 모드로 실행

  1"""plan_and_execute.py - Plan-and-Execute 완성본 (checkpoint 배포)"""
  2from typing import TypedDict, Annotated, Literal
  3from operator import add
  4
  5from dotenv import load_dotenv
  6from pydantic import BaseModel, Field
  7from langchain_aws import ChatBedrockConverse
  8from langgraph.graph import StateGraph, START, END
  9
 10load_dotenv()
 11
 12llm = ChatBedrockConverse(
 13    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
 14    region_name="us-east-1",
 15    temperature=0,
 16)
 17
 18
 19def get_text(message):
 20    """ChatBedrockConverse는 content를 블록 리스트로 주기도 하므로 텍스트만 모아 반환합니다."""
 21    content = message.content
 22    if isinstance(content, list):
 23        return "".join(block.get("text", "") for block in content if isinstance(block, dict))
 24    return content
 25
 26
 27# ---------- 1) State ----------
 28class State(TypedDict):
 29    user_request: str
 30    plan: list[str]
 31    past_steps: Annotated[list[tuple[str, str]], add]
 32    final_response: str
 33
 34
 35# ---------- 2) 계획 수립 노드 ----------
 36class Plan(BaseModel):
 37    steps: list[str] = Field(description="순서대로 실행할 단계 리스트")
 38
 39
 40planner_llm = llm.with_structured_output(Plan)
 41
 42
 43def make_plan(state: State) -> dict:
 44    plan = planner_llm.invoke(
 45        f"다음 요청을 5-7단계의 구체적 계획으로 분해하세요.\n"
 46        f"각 단계는 한 문장, 실행 가능한 행동이어야 합니다.\n\n"
 47        f"요청: {state['user_request']}"
 48    )
 49    print(f"[planner] 계획 {len(plan.steps)}단계")
 50    for i, s in enumerate(plan.steps):
 51        print(f"  {i + 1}. {s}")
 52    return {"plan": plan.steps}
 53
 54
 55# ---------- 3) 단계 실행 노드 ----------
 56def execute_step(state: State) -> dict:
 57    current = state["plan"][0]
 58    print(f"\n[execute] {current}")
 59
 60    context = (
 61        "전체 계획:\n"
 62        + "\n".join(f"  {i + 1}. {s}" for i, s in enumerate(state["plan"]))
 63        + "\n\n완료된 단계:\n"
 64        + "\n".join(f"  - {s}: {r[:60]}" for s, r in state["past_steps"])
 65        + f"\n\n지금 실행할 단계: {current}\n실행 결과를 한두 문장으로 요약하세요."
 66    )
 67    result = get_text(llm.invoke(context))
 68    return {
 69        "plan": state["plan"][1:],           # 첫 단계 제거 (덮어쓰기)
 70        "past_steps": [(current, result)],   # add 리듀서가 누적
 71    }
 72
 73
 74# ---------- 4) 재계획 노드 ----------
 75class Replan(BaseModel):
 76    """계획을 계속할지, 수정할지, 끝낼지 결정한다."""
 77    action: Literal["continue", "revise", "finish"] = Field(
 78        description="continue=남은 계획 유지, revise=계획 수정, finish=종료"
 79    )
 80    revised_plan: list[str] = Field(
 81        default_factory=list, description="revise일 때 새 남은 계획"
 82    )
 83    reason: str = Field(description="판단 근거 한 문장")
 84
 85
 86replanner_llm = llm.with_structured_output(Replan)
 87
 88
 89def replan(state: State) -> dict:
 90    decision = replanner_llm.invoke(
 91        f"원래 요청: {state['user_request']}\n\n"
 92        f"완료된 단계와 결과:\n"
 93        + "\n".join(f"  - {s}: {r[:80]}" for s, r in state["past_steps"])
 94        + "\n\n남은 계획:\n"
 95        + "\n".join(f"  - {s}" for s in state["plan"])
 96        + "\n\n판단하세요.\n"
 97          "- 남은 계획이 여전히 유효하면 continue\n"
 98          "- 직전 결과 때문에 남은 계획을 바꿔야 하면 revise (새 계획 제시)\n"
 99          "- 요청에 답하기 충분한 정보가 모였거나 남은 계획이 없으면 finish"
100    )
101    print(f"[replan] {decision.action} - {decision.reason}")
102    if decision.action == "revise":
103        return {"plan": decision.revised_plan}
104    if decision.action == "finish":
105        return {"plan": []}
106    return {}
107
108
109def after_replan(state: State) -> Literal["execute", "finalize"]:
110    return "finalize" if not state["plan"] else "execute"
111
112
113# ---------- 5) 마무리 노드 ----------
114def finalize(state: State) -> dict:
115    summary = "\n".join(f"- {s}: {r[:100]}" for s, r in state["past_steps"])
116    response = llm.invoke(
117        f"원래 요청: {state['user_request']}\n\n실행 내역:\n{summary}\n\n"
118        f"실행 내역을 근거로 최종 답변을 작성하세요."
119    )
120    return {"final_response": get_text(response)}
121
122
123# ---------- 6) 그래프 조립 ----------
124builder = StateGraph(State)
125builder.add_node("planner", make_plan)
126builder.add_node("execute", execute_step)
127builder.add_node("replan", replan)
128builder.add_node("finalize", finalize)
129
130builder.add_edge(START, "planner")
131builder.add_edge("planner", "execute")
132builder.add_edge("execute", "replan")
133builder.add_conditional_edges("replan", after_replan,
134                              {"execute": "execute", "finalize": "finalize"})
135builder.add_edge("finalize", END)
136
137graph = builder.compile()
138
139
140if __name__ == "__main__":
141    result = graph.invoke(
142        {
143            # 교육용 더미 데이터 기반 시나리오입니다
144            "user_request": "클라우드운영팀의 이번 분기 클라우드 비용을 분석해서 다음 분기 예산을 추정",
145            "plan": [],
146            "past_steps": [],
147            "final_response": "",
148        },
149        {"recursion_limit": 40},   # 안전장치: 무한 반복 방지
150    )
151    print(f"\n=== 최종 답변 ===\n{result['final_response']}")

 1"""long_term_memory.py - Store 저장, 회수, 그래프 연결 (checkpoint 배포)"""
 2from dotenv import load_dotenv
 3from langchain_aws import ChatBedrockConverse
 4from langchain_core.messages import SystemMessage, HumanMessage
 5from langgraph.store.memory import InMemoryStore
 6from langgraph.checkpoint.memory import InMemorySaver
 7from langgraph.graph import StateGraph, START, END, MessagesState
 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
16store = InMemoryStore()
17user_id = "haneul_kim"   # Store 네임스페이스 라벨에는 점을 쓸 수 없습니다
18
19
20def get_text(message):
21    """ChatBedrockConverse는 content를 블록 리스트로 주기도 하므로 텍스트만 모아 반환합니다."""
22    content = message.content
23    if isinstance(content, list):
24        return "".join(block.get("text", "") for block in content if isinstance(block, dict))
25    return content
26
27
28# ---------- 1) 저장 ----------
29store.put(
30    ("users", user_id),          # namespace: 사용자별로 분리
31    "profile",                   # key
32    {                            # value: 딕셔너리 (교육용 더미 데이터)
33        "name": "김하늘",
34        "team": "클라우드운영팀",
35        "email": "haneul.kim@samsungsds.example.com",
36        "preferences": "답변은 짧고 핵심 위주로, 코드 예시는 Python 기준",
37        "rules": "외부 발송 작업은 반드시 승인 절차를 거친다",
38    },
39)
40
41# ---------- 2) 회수 ----------
42item = store.get(("users", user_id), "profile")
43print("[get]", item.value)
44
45# 검색(search): namespace 안에서 조건이나 유사도로 찾기
46results = store.search(("users", user_id))
47for r in results:
48    print("[search]", r.key, "->", r.value["name"])
49
50
51# ---------- 3) 그래프에 연결 ----------
52def chat_node(state: MessagesState, config, *, store):
53    uid = config["configurable"]["user_id"]
54    profile = store.get(("users", uid), "profile")
55    memory_text = str(profile.value) if profile else "저장된 기억 없음"
56
57    messages = [
58        SystemMessage(content=f"[사용자 장기 기억]\n{memory_text}\n"
59                              f"기억을 반영해 답하세요."),
60        *state["messages"],
61    ]
62    return {"messages": [llm.invoke(messages)]}
63
64
65builder = StateGraph(MessagesState)
66builder.add_node("chat", chat_node)
67builder.add_edge(START, "chat")
68builder.add_edge("chat", END)
69
70# compile에 checkpointer(단기)와 store(장기)를 함께 전달합니다
71graph = builder.compile(checkpointer=InMemorySaver(), store=store)
72
73# ---------- 4) 서로 다른 세션에서 기억이 유지되는지 확인 ----------
74for thread in ["thread-1", "thread-2"]:   # thread가 달라도 (= 새 세션)
75    config = {"configurable": {"thread_id": thread, "user_id": user_id}}
76    out = graph.invoke(
77        {"messages": [HumanMessage(content="보고서 메일 초안 쓸 때 주의할 점 알려줘")]},
78        config,
79    )
80    print(f"\n[{thread}] {get_text(out['messages'][-1])[:]}")

  1# final_scenario.py - 6일차 종합: RAG + MCP + Supervisor
  2import asyncio
  3import sys
  4from dotenv import load_dotenv
  5from langchain_aws import ChatBedrockConverse, BedrockEmbeddings
  6from langchain_chroma import Chroma
  7from langchain_core.tools import tool
  8from langchain_core.messages import HumanMessage
  9from langchain_mcp_adapters.client import MultiServerMCPClient
 10from langgraph.prebuilt import create_react_agent
 11from langgraph_supervisor import create_supervisor
 12
 13load_dotenv()
 14MODEL = "us.anthropic.claude-sonnet-4-5-20250929-v1:0"
 15
 16def get_text(message):
 17    """ChatBedrockConverse는 content를 블록 리스트로 주기도 하므로 텍스트만 모아 반환합니다."""
 18    content = message.content
 19    if isinstance(content, list):
 20        return "".join(block.get("text", "") for block in content if isinstance(block, dict))
 21    return content
 22
 23def make_worker_llm():
 24    return ChatBedrockConverse(model=MODEL, region_name="us-east-1", temperature=0)
 25
 26# ---------- research_agent: 2일차 RAG ----------
 27embeddings = BedrockEmbeddings(
 28    model_id="amazon.titan-embed-text-v2:0",
 29    region_name="us-east-1",
 30)
 31vectorstore = Chroma(
 32    persist_directory="./chroma_db",
 33    embedding_function=embeddings,
 34    collection_name="sds_policies",
 35)
 36retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
 37
 38@tool
 39def retrieve_docs(query: str) -> str:
 40    """삼성SDS 사내 규정 문서(연차, 재택근무, 출장 등)에서 관련 내용을 검색한다."""
 41    docs = retriever.invoke(query)
 42    if not docs:
 43        return "관련 문서를 찾지 못했습니다."
 44    return "\n\n".join(
 45        f"[출처: {d.metadata.get('source', '알 수 없음')}]\n{d.page_content}"
 46        for d in docs
 47    )
 48
 49research_agent = create_react_agent(
 50    make_worker_llm(),
 51    [retrieve_docs],
 52    prompt=(
 53        "너는 삼성SDS 사내 규정 문서 검색 전문가다.\n"
 54        "반드시 retrieve_docs 결과에 근거해 답하고 출처를 표기하라.\n"
 55        "담당 범위 밖 질문에는 '담당 범위가 아닙니다'라고만 답하라."
 56    ),
 57    name="research_agent",
 58)
 59
 60general_agent = create_react_agent(
 61    make_worker_llm(),
 62    [],
 63    prompt=(
 64        "너는 일반 응대 담당이다. 인사말과 잡담에 친절히 답하고,\n"
 65        "업무 질문은 '사내 데이터 조회와 규정 검색을 도와드릴 수 있습니다'로 안내하라."
 66    ),
 67    name="general_agent",
 68)
 69
 70async def build_app():
 71    # ---------- data_agent: 4일차 MCP ----------
 72    client = MultiServerMCPClient(
 73        {
 74            "company": {
 75                "command": sys.executable,
 76                "args": ["mcp_server.py"],
 77                "transport": "stdio",
 78            }
 79        }
 80    )
 81    mcp_tools = await client.get_tools()
 82
 83    data_agent = create_react_agent(
 84        make_worker_llm(),
 85        mcp_tools,
 86        prompt=(
 87            "너는 삼성SDS 사내 데이터 조회 전문가다. 임직원, 자산, 프로젝트 조회만 담당한다.\n"
 88            "개인은 search_employee, 팀 명단과 인원수는 list_team_members를 쓴다.\n"
 89            "반드시 도구로 조회한 결과에 근거해 답하고, 한 문단으로 정리해 보고하라.\n"
 90            "담당 범위 밖 질문에는 '담당 범위가 아닙니다'라고만 답하라."
 91        ),
 92        name="data_agent",
 93    )
 94
 95    # ---------- Supervisor 조립 ----------
 96    supervisor = create_supervisor(
 97        [data_agent, research_agent, general_agent],
 98        model=ChatBedrockConverse(model=MODEL, region_name="us-east-1", temperature=0),
 99        prompt=(
100            "너는 삼성SDS 사내 어시스턴트의 Supervisor다.\n"
101            "[배분 기준]\n"
102            "- 임직원, 자산, 프로젝트 등 사내 시스템 데이터 조회는 data_agent\n"
103            "- 연차, 재택근무, 출장 등 사내 규정 문서 검색은 research_agent\n"
104            "- 그 외 인사말, 잡담은 general_agent\n"
105            "[규칙]\n"
106            "- 직접 답을 지어내지 말고 반드시 담당 Agent를 통해 확인하라.\n"
107            "- 여러 요구가 섞인 질문은 각 Agent를 순서대로 호출해 모두 처리하라.\n"
108            "- 마지막에는 문서 근거와 데이터를 함께 담아 하나의 보고로 종합하고,\n"
109            "  결론(가능, 불가능, 조건부)을 명시하라."
110        ),
111    )
112    return supervisor.compile()
113
114async def main():
115    app = await build_app()
116    question = (
117        "출장 규정에서 식비와 숙박비 근거를 찾고, 물류플랫폼팀 임직원 수와 "
118        "진행 중인 프로젝트를 조회해서, 다음 분기 출장 예산 검토 보고를 종합해줘"
119    )
120    print(f"질문: {question}\n")
121
122    async for event in app.astream(
123        {"messages": [HumanMessage(content=question)]},
124        stream_mode="updates",
125        config={"recursion_limit": 30},
126    ):
127        for node, update in event.items():
128            for m in (update or {}).get("messages", []):
129                if getattr(m, "tool_calls", None):
130                    for tc in m.tool_calls:
131                        print(f"[{node}] 도구: {tc['name']}({tc.get('args', {})})")
132                elif getattr(m, "content", None):
133                    label = getattr(m, "name", None) or node
134                    print(f"[{label}] {get_text(m)[:200]}")
135
136if __name__ == "__main__":
137    asyncio.run(main())

  1# final_scenario.py - 6일차 종합: RAG + MCP + Supervisor
  2import asyncio
  3import sys
  4from dotenv import load_dotenv
  5from langchain_aws import ChatBedrockConverse, BedrockEmbeddings
  6from langchain_chroma import Chroma
  7from langchain_core.tools import tool
  8from langchain_core.messages import HumanMessage
  9from langchain_mcp_adapters.client import MultiServerMCPClient
 10from langgraph.prebuilt import create_react_agent
 11from langgraph_supervisor import create_supervisor
 12from langgraph.checkpoint.memory import InMemorySaver
 13from langgraph.store.memory import InMemoryStore
 14
 15load_dotenv()
 16MODEL = "global.anthropic.claude-haiku-4-5-20251001-v1:0"
 17
 18def get_text(message):
 19    """ChatBedrockConverse는 content를 블록 리스트로 주기도 하므로 텍스트만 모아 반환합니다."""
 20    content = message.content
 21    if isinstance(content, list):
 22        return "".join(block.get("text", "") for block in content if isinstance(block, dict))
 23    return content
 24
 25def make_worker_llm():
 26    return ChatBedrockConverse(model=MODEL, region_name="us-east-1", temperature=0)
 27
 28# ---------- research_agent: 2일차 RAG ----------
 29embeddings = BedrockEmbeddings(
 30    model_id="amazon.titan-embed-text-v2:0",
 31    region_name="us-east-1",
 32)
 33vectorstore = Chroma(
 34    persist_directory="./chroma_db",
 35    embedding_function=embeddings,
 36    collection_name="sds_policies",
 37)
 38retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
 39
 40@tool
 41def retrieve_docs(query: str) -> str:
 42    """삼성SDS 사내 규정 문서(연차, 재택근무, 출장 등)에서 관련 내용을 검색한다."""
 43    docs = retriever.invoke(query)
 44    if not docs:
 45        return "관련 문서를 찾지 못했습니다."
 46    return "\n\n".join(
 47        f"[출처: {d.metadata.get('source', '알 수 없음')}]\n{d.page_content}"
 48        for d in docs
 49    )
 50
 51research_agent = create_react_agent(
 52    make_worker_llm(),
 53    [retrieve_docs],
 54    prompt=(
 55        "너는 삼성SDS 사내 규정 문서 검색 전문가다.\n"
 56        "반드시 retrieve_docs 결과에 근거해 답하고 출처를 표기하라.\n"
 57        "담당 범위 밖 질문에는 '담당 범위가 아닙니다'라고만 답하라."
 58    ),
 59    name="research_agent",
 60)
 61
 62general_agent = create_react_agent(
 63    make_worker_llm(),
 64    [],
 65    prompt=(
 66        "너는 일반 응대 담당이다. 인사말과 잡담에 친절히 답하고,\n"
 67        "업무 질문은 '사내 데이터 조회와 규정 검색을 도와드릴 수 있습니다'로 안내하라."
 68    ),
 69    name="general_agent",
 70)
 71
 72async def build_app():
 73    # ---------- data_agent: 4일차 MCP ----------
 74    client = MultiServerMCPClient(
 75        {
 76            "company": {
 77                "command": sys.executable,
 78                "args": ["mcp_server.py"],
 79                "transport": "stdio",
 80            }
 81        }
 82    )
 83    mcp_tools = await client.get_tools()
 84
 85    data_agent = create_react_agent(
 86        make_worker_llm(),
 87        mcp_tools,
 88        prompt=(
 89            "너는 삼성SDS 사내 데이터 조회 전문가다. 임직원, 자산, 프로젝트 조회만 담당한다.\n"
 90            "개인은 search_employee, 팀 명단과 인원수는 list_team_members를 쓴다.\n"
 91            "반드시 도구로 조회한 결과에 근거해 답하고, 한 문단으로 정리해 보고하라.\n"
 92            "담당 범위 밖 질문에는 '담당 범위가 아닙니다'라고만 답하라."
 93        ),
 94        name="data_agent",
 95    )
 96
 97    # ---------- Supervisor 조립 ----------
 98    supervisor = create_supervisor(
 99        [data_agent, research_agent, general_agent],
100        model=ChatBedrockConverse(model=MODEL, region_name="us-east-1", temperature=0),
101        prompt=(
102            "너는 삼성SDS 사내 어시스턴트의 Supervisor다.\n"
103            "[배분 기준]\n"
104            "- 임직원, 자산, 프로젝트 등 사내 시스템 데이터 조회는 data_agent\n"
105            "- 연차, 재택근무, 출장 등 사내 규정 문서 검색은 research_agent\n"
106            "- 그 외 인사말, 잡담은 general_agent\n"
107            "[규칙]\n"
108            "- 직접 답을 지어내지 말고 반드시 담당 Agent를 통해 확인하라.\n"
109            "- 여러 요구가 섞인 질문은 각 Agent를 순서대로 호출해 모두 처리하라.\n"
110            "- 마지막에는 문서 근거와 데이터를 함께 담아 하나의 보고로 종합하고,\n"
111            "  결론(가능, 불가능, 조건부)을 명시하라."
112        ),
113    )
114    checkpointer = InMemorySaver()   # 단기 기억: thread_id 단위 대화 흐름
115    store = InMemoryStore()          # 장기 기억: user_id 단위 사용자 기억
116    return supervisor.compile(checkpointer=checkpointer, store=store)
117
118async def main():
119    app = await build_app()
120    question = (
121        "출장 규정에서 식비와 숙박비 근거를 찾고, 물류플랫폼팀 임직원 수와 "
122        "진행 중인 프로젝트를 조회해서, 다음 분기 출장 예산 검토 보고를 종합해줘"
123    )
124    print(f"질문: {question}\n")
125
126    async for event in app.astream(
127        {"messages": [HumanMessage(content=question)]},
128        stream_mode="updates",
129        config={"recursion_limit": 30,
130        "configurable": {"thread_id": "session-1", "user_id": "haneul_kim"}},
131    ):
132        for node, update in event.items():
133            for m in (update or {}).get("messages", []):
134                if getattr(m, "tool_calls", None):
135                    for tc in m.tool_calls:
136                        print(f"[{node}] 도구: {tc['name']}({tc.get('args', {})})")
137                elif getattr(m, "content", None):
138                    label = getattr(m, "name", None) or node
139                    print(f"[{label}] {get_text(m)[:200]}")
140
141if __name__ == "__main__":
142    asyncio.run(main())

 1"""local_tracer.py - 실행 기록을 JSONL 파일로 남기는 콜백 핸들러"""
 2import json
 3import time
 4from pathlib import Path
 5from langchain_core.callbacks import BaseCallbackHandler
 6
 7
 8def get_text(message):
 9    """ChatBedrockConverse는 content를 블록 리스트로 주기도 하므로 텍스트만 모아 반환합니다."""
10    content = message.content
11    if isinstance(content, list):
12        return "".join(block.get("text", "") for block in content if isinstance(block, dict))
13    return content
14
15
16class FileTracer(BaseCallbackHandler):
17    """LLM 호출, 도구 호출을 JSONL 한 줄씩 기록한다."""
18
19    def __init__(self, path: str = "trace.jsonl"):
20        self.path = Path(path)
21        self._starts = {}   # run_id -> 시작 시각
22
23    def _write(self, record: dict):
24        record["ts"] = time.strftime("%Y-%m-%d %H:%M:%S")
25        with self.path.open("a", encoding="utf-8") as f:
26            f.write(json.dumps(record, ensure_ascii=False) + "\n")
27
28    # ---- LLM ----
29    def on_chat_model_start(self, serialized, messages, *, run_id, **kw):
30        self._starts[run_id] = time.time()
31        flat = [get_text(m)[:200] for batch in messages for m in batch]
32        self._write({"event": "llm_start", "run_id": str(run_id), "input": flat})
33
34    def on_llm_end(self, response, *, run_id, **kw):
35        elapsed = time.time() - self._starts.pop(run_id, time.time())
36        usage = {}
37        gen = response.generations[0][0]
38        msg = getattr(gen, "message", None)
39        if msg is not None and getattr(msg, "usage_metadata", None):
40            usage = msg.usage_metadata     # 입력, 출력 토큰 수
41        self._write({
42            "event": "llm_end", "run_id": str(run_id),
43            "output": gen.text[:200], "latency_s": round(elapsed, 2),
44            "usage": usage,
45        })
46
47    # ---- 도구 ----
48    def on_tool_start(self, serialized, input_str, *, run_id, **kw):
49        self._starts[run_id] = time.time()
50        self._write({"event": "tool_start", "run_id": str(run_id),
51                     "tool": serialized.get("name"), "input": input_str[:200]})
52
53    def on_tool_end(self, output, *, run_id, **kw):
54        elapsed = time.time() - self._starts.pop(run_id, time.time())
55        self._write({"event": "tool_end", "run_id": str(run_id),
56                     "output": str(output)[:200], "latency_s": round(elapsed, 2)})
57
58    def on_llm_error(self, error, *, run_id, **kw):
59        self._write({"event": "llm_error", "run_id": str(run_id), "error": str(error)})