되돌아갈 줄 아는 그래프, LangGraph

13def retrieve(question: str) -> dict:
14    """2일차 retriever 자리입니다.
15    진짜 벡터 검색 대신, 질문에 '재택'이라는 말이 있을 때만
16    문서를 찾았다고 치는 가짜 검색기입니다."""
17    if "재택" in question:
18        found = ["재택근무는 주 2회까지 가능하다."]   # 검색 성공
19    else:
20        found = []                                    # 검색 실패: 관련 문서를 못 찾음
21    print(f"  [retrieve] 질문={question!r} -> 문서 {len(found)}건")
22    return {"question": question, "docs": found}
23
24
25def build_prompt(data: dict) -> dict:
26    """2일차 prompt 자리입니다. 찾은 문서를 문자열 하나로 합칩니다."""
27    context = "\n".join(data["docs"]) if data["docs"] else "(근거 문서 없음)"
28    return {"question": data["question"], "context": context}
29
30
31def generate(data: dict) -> str:
32    """2일차 llm 자리입니다. 근거가 없으면 모른다고 답하도록 흉내만 냅니다."""
33    if data["context"] == "(근거 문서 없음)":
34        return "죄송합니다. 관련 규정을 찾지 못했습니다."
35    return f"답변: {data['context']}"

38# ---------- 2) 부품을 체인으로 잇습니다. 2일차와 같은 모양입니다 ----------
39# retrieve -> build_prompt -> generate 로 왼쪽에서 오른쪽으로 한 번 흐릅니다.
40chain = RunnableLambda(retrieve) | RunnableLambda(build_prompt) | RunnableLambda(generate)
41
42
43# ---------- 3) 검색이 실패하는 질문을 넣어 봅니다 ----------
44# '재택'이라는 말이 없으므로 retrieve가 빈손으로 돌아옵니다.
45print("[1] 체인을 한 번 실행합니다")
46print("  결과:", chain.invoke("집에서 일해도 되나요?"))

60question = "집에서 일해도 되나요?"
61alternatives = ["재택근무 규정", "재택근무 며칠"]   # 바꿔서 다시 물어볼 질문 후보
62attempt = 0
63
64answer = chain.invoke(question)                    # 첫 시도
65
66while "찾지 못했습니다" in answer and attempt < len(alternatives):
67    question = alternatives[attempt]               # 질문을 바꾸고
68    attempt += 1
69    print(f"  [바깥 루프] {attempt}번째 재시도 -> 질문을 {question!r} 로 바꿉니다")
70    answer = chain.invoke(question)                # 체인을 처음부터 다시 실행
71
72print(f"\n최종 답변: {answer}")
73print(f"재시도 횟수: {attempt}")

 1[1] 체인을 한 번 실행합니다
 2  [retrieve] 질문='집에서 일해도 되나요?' -> 문서 0건
 3  결과: 죄송합니다. 관련 규정을 찾지 못했습니다.
 4
 5[2] 재검색을 넣으려면 체인 바깥에 루프를 둬야 합니다
 6  [retrieve] 질문='집에서 일해도 되나요?' -> 문서 0건
 7  [바깥 루프] 1번째 재시도 -> 질문을 '재택근무 규정' 로 바꿉니다
 8  [retrieve] 질문='재택근무 규정' -> 문서 1건
 9
10최종 답변: 답변: 재택근무는 주 2회까지 가능하다.
11재시도 횟수: 1
12
13[관찰] 다시 찾을지 말지를 판단하는 코드와 반복하는 코드가
14       chain 안에 있습니까, 이 파일의 while 문 안에 있습니까?

 2from typing import TypedDict
 3
 4
 5class State(TypedDict):
 6    user_input: str   # 사용자가 넣는 입력
 7    response: str     # 그래프가 채울 응답
 8    search_quality: bool
 9
10
11initial: State = {"user_input": "안녕하세요", "response": "", "search_quality": False}
12print(initial)
13print(type(initial))   # 실행 시점에는 그냥 dict입니다

1{'user_input': '안녕하세요', 'response': '', 'search_quality': False}
2<class 'dict'>

 6class State(TypedDict):
 7    user_input: str
 8    word_count: int
 9    response: str
10
11
12def clean(state: State) -> dict:
13    """앞뒤 공백과 연속 공백을 정리합니다. 이미 있는 user_input을 덮어씁니다."""
14    print(f"  [clean]   받은 State: {state}")
15    cleaned = " ".join(state["user_input"].split())
16    return {"user_input": cleaned}
17
18
19def count(state: State) -> dict:
20    """단어 수를 세서 새 키 word_count를 추가합니다."""
21    print(f"  [count]   받은 State: {state}")
22    return {"word_count": len(state["user_input"].split())}
23
24
25def respond(state: State) -> dict:
26    """앞 노드들이 만든 값을 모아 답변을 만듭니다."""
27    print(f"  [respond] 받은 State: {state}")
28    user = state["user_input"]
29    return {"response": f"입력하신 내용은 '{user}' 이고, 단어 {state['word_count']}개입니다."}

32builder = StateGraph(State)
33builder.add_node("clean", clean)
34builder.add_node("count", count)
35builder.add_node("respond", respond)
36builder.add_edge(START, "clean")
37builder.add_edge("clean", "count")
38builder.add_edge("count", "respond")
39builder.add_edge("respond", END)
40
41graph = builder.compile()    # 그래프를 실행 가능한 객체로 변환

44if __name__ == "__main__":   # langgraph dev 서버가 import할 때는 실행되지 않도록 가드 아래에 둡니다
45    # invoke 전에는 State 값이 없습니다. State는 데이터 모양(타입)일 뿐이고, 값은 실행할 때 생깁니다
46    user_input = "  안녕하세요    랭그래프   "
47
48    # 1) 단계마다 State 전체 (노드 안의 print는 '받은 State', 여기는 '노드 실행 뒤 State')
49    print("[values] 단계마다 State 전체")
50    for i, step in enumerate(graph.stream({"user_input": user_input}, stream_mode="values")):
51        print(f"{i}단계 뒤 State: {step}\n")
52
53    # 2) 단계마다 노드가 돌려준 값만
54    print("[updates] 단계마다 노드가 돌려준 값만")
55    for step in graph.stream({"user_input": user_input}, stream_mode="updates"):
56        print(f"  돌려준 값: {step}")
57
58    # 3) invoke는 마지막 State만 돌려줍니다
59    print("\n[invoke] 최종 State")
60    result = graph.invoke({"user_input": user_input})
61    print(result)

 1[values] 단계마다 State 전체
 20단계 뒤 State: {'user_input': '  안녕하세요    랭그래프   '}
 3
 4  [clean]   받은 State: {'user_input': '  안녕하세요    랭그래프   '}
 51단계 뒤 State: {'user_input': '안녕하세요 랭그래프'}
 6
 7  [count]   받은 State: {'user_input': '안녕하세요 랭그래프'}
 82단계 뒤 State: {'user_input': '안녕하세요 랭그래프', 'word_count': 2}
 9
10  [respond] 받은 State: {'user_input': '안녕하세요 랭그래프', 'word_count': 2}
113단계 뒤 State: {'user_input': '안녕하세요 랭그래프', 'word_count': 2, 'response': "입력하신 내용은 '안녕하세요 랭그래프' 이고, 단어 2개입니다."}
12
13[updates] 단계마다 노드가 돌려준 값만
14  [clean]   받은 State: {'user_input': '  안녕하세요    랭그래프   '}
15  돌려준 값: {'clean': {'user_input': '안녕하세요 랭그래프'}}
16  [count]   받은 State: {'user_input': '안녕하세요 랭그래프'}
17  돌려준 값: {'count': {'word_count': 2}}
18  [respond] 받은 State: {'user_input': '안녕하세요 랭그래프', 'word_count': 2}
19  돌려준 값: {'respond': {'response': "입력하신 내용은 '안녕하세요 랭그래프' 이고, 단어 2개입니다."}}
20
21[invoke] 최종 State
22  [clean]   받은 State: {'user_input': '  안녕하세요    랭그래프   '}
23  [count]   받은 State: {'user_input': '안녕하세요 랭그래프'}
24  [respond] 받은 State: {'user_input': '안녕하세요 랭그래프', 'word_count': 2}
25{'user_input': '안녕하세요 랭그래프', 'word_count': 2, 'response': "입력하신 내용은 '안녕하세요 랭그래프' 이고, 단어 2개입니다."}

63    # 구조 시각화: PNG 저장이 안 되는 환경도 있으므로 try/except로 감쌉니다
64    try:
65        graph.get_graph().draw_mermaid_png(output_file_path="hello_graph.png")
66        print("\nhello_graph.png 저장 완료")
67    except Exception as e:
68        print("\nPNG 저장은 건너뜁니다:", e)
69        print(graph.get_graph().draw_mermaid())   # 텍스트 출력으로 대체

 8class ChatState(TypedDict):
 9    messages: Annotated[list, add_messages]
10
11
12def chat(state: ChatState) -> dict:
13    # 새 메시지를 리스트로 반환만 하면 add_messages가 알아서 누적합니다
14    return {"messages": [AIMessage(content="안녕하세요, 반갑습니다.")]}
15
16
17builder = StateGraph(ChatState)
18builder.add_node("chat", chat)
19builder.add_edge(START, "chat")
20builder.add_edge("chat", END)
21graph = builder.compile()

24if __name__ == "__main__":   # langgraph dev 서버가 import할 때는 실행되지 않도록 가드 아래에 둡니다
25    # 1차 실행
26    result = graph.invoke({"messages": [HumanMessage(content="안녕하세요?")]})
27    print("1차 실행 후 메시지 개수:", len(result["messages"]))
28
29    # 2차 실행: 이전 messages를 그대로 이어서 넘기면 계속 누적됩니다
30    result2 = graph.invoke(
31        {"messages": result["messages"] + [HumanMessage(content="잘 지내시나요?")]}
32    )
33    print("2차 실행 후 메시지 개수:", len(result2["messages"]))
34    for m in result2["messages"]:
35        print(f"  [{type(m).__name__}] {m.content}")

11차 실행 후 메시지 개수: 2
22차 실행 후 메시지 개수: 4
3  [HumanMessage] 안녕하세요?
4  [AIMessage] 안녕하세요, 반갑습니다.
5  [HumanMessage] 잘 지내시나요?
6  [AIMessage] 안녕하세요, 반갑습니다.

52def answer_question(state: ClassifyState) -> dict:
53    """질문에 LLM으로 답합니다."""
54    response = llm.invoke(f"다음 질문에 한 줄로 답하세요: {state['user_input']}")
55    return {"response": get_text(response)}
56
57
58def react_exclaim(state: ClassifyState) -> dict:
59    """감탄에는 LLM 없이 정해진 반응을 합니다."""
60    return {"response": f"'{state['user_input']}' 라니 멋진 일이네요."}
61
62
63def route_by_intent(state: ClassifyState) -> Literal["answer_question", "react_exclaim"]:
64    """라우팅 함수: State를 보고 다음 노드의 이름을 반환합니다."""
65    if state["intent"] == "question":
66        return "answer_question"
67    return "react_exclaim"

70builder = StateGraph(ClassifyState)
71builder.add_node("classify", classify)
72builder.add_node("answer_question", answer_question)
73builder.add_node("react_exclaim", react_exclaim)
74
75builder.add_edge(START, "classify")
76builder.add_conditional_edges("classify", route_by_intent)   # 분기 등록
77builder.add_edge("answer_question", END)
78builder.add_edge("react_exclaim", END)
79graph = builder.compile()

15# 분류 결과 스키마 - Literal로 두 값 중 하나만 나오도록 강제합니다
16class Intent(BaseModel):
17    intent: Literal["question", "exclaim"] = Field(
18        description="답을 원하는 질문이면 'question', 감탄이나 소감이면 'exclaim'"
19    )
20
21
22# 분류는 매번 같은 답이 나와야 하므로 temperature=0으로 따로 둡니다
23classifier = ChatBedrockConverse(
24    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
25    region_name="us-east-1",
26    temperature=0,
27).with_structured_output(Intent)

38class ClassifyState(TypedDict):
39    user_input: str
40    intent: str
41    response: str
42
43
44def classify(state: ClassifyState) -> dict:
45    """LLM이 입력을 보고 question / exclaim 중 하나로 분류합니다. 물음표 유무와 무관합니다."""
46    result = classifier.invoke(
47        f"다음 문장이 답을 원하는 질문인지, 감탄이나 소감인지 분류하세요: {state['user_input']}"
48    )
49    return {"intent": result.intent}

82if __name__ == "__main__":   # langgraph dev 서버가 import할 때는 실행되지 않도록 가드 아래에 둡니다
83    for text in [
84        "오늘 사내 식당 메뉴 뭐야?",
85        "와 배포가 한 번에 성공했다",
86        "LangGraph가 뭐야?",
87        "오늘 메뉴 뭐야",         # 물음표 없는 질문 -> 이전 규칙에서는 exclaim으로 오분류
88        "이게 되네?!",            # 물음표 있는 감탄 -> 이전 규칙에서는 question으로 오분류
89    ]:
90        result = graph.invoke({"user_input": text, "intent": "", "response": ""})
91        print(f"입력: {text}")
92        print(f"분류: {result['intent']} / 응답: {result['response']}\n")

22class QualityState(TypedDict):
23    question: str
24    answer: str
25    attempts: int
26    quality_score: float
27
28
29def generate(state: QualityState) -> dict:
30    """답변을 생성하고 시도 횟수를 1 올립니다."""
31    response = llm.invoke(
32        f"다음 질문에 200자 이상으로 자세하게 답하세요: {state['question']}"
33    )
34    return {
35        "answer": get_text(response),
36        "attempts": state.get("attempts", 0) + 1,
37    }
38
39
40def check_quality(state: QualityState) -> dict:
41    """품질 평가. 실습용으로 단순하게 길이 기반 점수를 씁니다 (200자면 만점)."""
42    score = min(len(state["answer"]) / 500, 1.0)
43    return {"quality_score": score}

46def route_quality(state: QualityState) -> str:
47    """품질이 좋거나 시도 한도에 도달하면 종료하고 아니면 재시도합니다."""
48    if state["quality_score"] >= 0.8:
49        return END
50    if state["attempts"] >= 10:       # 안전장치: 무한 루프 방지
51        return END
52    return "generate"                # 이전 노드로 돌아감 = 반복
53
54
55builder = StateGraph(QualityState)
56builder.add_node("generate", generate)
57builder.add_node("check_quality", check_quality)
58builder.add_edge(START, "generate")
59builder.add_edge("generate", "check_quality")
60builder.add_conditional_edges("check_quality", route_quality)
61graph = builder.compile()

63if __name__ == "__main__":   # 5번째 세션에서 이 파일을 import해 쓰므로 실행 부분은 가드 아래에 둡니다
64    result = {
65        "question": "LangGraph의 장점을 한문장으로 설명해 주세요",
66        "answer": "",
67        "attempts": 0,
68        "quality_score": 0.0,
69    }
70
71    # invoke 대신 stream으로 돌려 노드가 끝날 때마다 중간 과정을 출력합니다
72    # updates 모드는 노드가 돌려준 값만 주므로 result에 직접 합쳐 최종 State를 만듭니다
73    for step in graph.stream(result, stream_mode="updates"):
74        node, update = next(iter(step.items()))
75        result.update(update)
76        if node == "generate":
77            print(f"[{result['attempts']}회차] generate      -> 답변 {len(update['answer'])}자")
78        elif node == "check_quality":
79            score = update["quality_score"]
80            if score >= 0.8:
81                verdict = "통과, 종료"
82            elif route_quality(result) == END:
83                verdict = "미달이지만 시도 한도, 종료"
84            else:
85                verdict = "미달, 재시도"
86            print(f"[{result['attempts']}회차] check_quality -> 점수 {score:.2f} ({verdict})")

 1[1회차] generate      -> 답변 300자
 2[1회차] check_quality -> 점수 0.60 (미달, 재시도)
 3[2회차] generate      -> 답변 450자
 4[2회차] check_quality -> 점수 0.90 (통과, 종료)
 5
 6=== 최종 결과 ===
 7시도 횟수: 2
 8품질 점수: 0.90
 9답변 길이: 450자
10답변 미리보기: ...

14class State(TypedDict):
15    question: str
16    alternatives: list[str]   # 바꿔서 다시 물어볼 질문 후보
17    attempt: int              # 재시도 횟수
18    docs: list[str]
19    answer: str

24def retrieve(state: State) -> dict:
25    """가짜 검색기: 질문에 '재택'이 있을 때만 문서를 찾습니다."""
26    q = state["question"]
27    docs = ["재택근무는 주 2회까지 가능하다."] if "재택" in q else []
28    print(f"  [retrieve] 질문={q!r} -> 문서 {len(docs)}건")
29    return {"docs": docs}
30
31
32def rewrite(state: State) -> dict:
33    """질문을 다음 후보로 바꾸고 재시도 횟수를 올립니다."""
34    attempt = state["attempt"]
35    new_q = state["alternatives"][attempt]
36    print(f"  [rewrite]  {attempt + 1}번째 재시도 -> 질문을 {new_q!r} 로 바꿉니다")
37    return {"question": new_q, "attempt": attempt + 1}
38
39
40def generate(state: State) -> dict:
41    print("  [generate] 문서로 답변을 만듭니다")
42    return {"answer": f"답변: {state['docs'][0]}"}
43
44
45def give_up(state: State) -> dict:
46    print("  [give_up]  재시도를 다 써서 포기합니다")
47    return {"answer": "죄송합니다. 관련 규정을 찾지 못했습니다."}
48
49
50# ---------- 분기 함수: 다음에 갈 노드 이름을 돌려줍니다 ----------
51
52def route_after_retrieve(state: State) -> str:
53    if state["docs"]:
54        return "generate"
55    if state["attempt"] < len(state["alternatives"]):
56        return "rewrite"
57    return "give_up"

62builder = StateGraph(State)
63builder.add_node("retrieve", retrieve)
64builder.add_node("rewrite", rewrite)
65builder.add_node("generate", generate)
66builder.add_node("give_up", give_up)
67
68builder.add_edge(START, "retrieve")
69builder.add_conditional_edges(                 # 분기: State를 보고 셋 중 하나로
70    "retrieve",
71    route_after_retrieve,
72    ["generate", "rewrite", "give_up"],
73)
74builder.add_edge("rewrite", "retrieve")        # 루프: 다시 검색으로 되돌아감
75builder.add_edge("generate", END)
76builder.add_edge("give_up", END)
77
78graph = builder.compile()

 84    cases = [
 85        ("바로 찾음",       "재택근무 며칠 가능해?", ["재택근무 규정"]),
 86        ("두 번 돌고 찾음", "집에서 일해도 되나요?", ["원격 근무 규정", "재택근무 규정"]),
 87        ("끝내 못 찾음",    "집에서 일해도 되나요?", ["원격 근무 규정", "홈오피스 규정"]),
 88    ]
 89
 90    for title, question, alternatives in cases:
 91        print("=" * 60)
 92        print(f"[{title}] 질문={question!r}, 후보={alternatives}")
 93        path = []
 94        final = {}
 95        for step in graph.stream(
 96            {"question": question, "alternatives": alternatives, "attempt": 0, "docs": [], "answer": ""},
 97            stream_mode="updates",
 98        ):
 99            node, update = next(iter(step.items()))
100            path.append(node)
101            final.update(update)
102        print(f"  지나간 경로: START -> {' -> '.join(path)} -> END")
103        print(f"  최종 답변: {final['answer']}")

 1============================================================
 2[바로 찾음] 질문='재택근무 며칠 가능해?', 후보=['재택근무 규정']
 3  [retrieve] 질문='재택근무 며칠 가능해?' -> 문서 1건
 4  [generate] 문서로 답변을 만듭니다
 5  지나간 경로: START -> retrieve -> generate -> END
 6  최종 답변: 답변: 재택근무는 주 2회까지 가능하다.
 7============================================================
 8[두 번 돌고 찾음] 질문='집에서 일해도 되나요?', 후보=['원격 근무 규정', '재택근무 규정']
 9  [retrieve] 질문='집에서 일해도 되나요?' -> 문서 0건
10  [rewrite]  1번째 재시도 -> 질문을 '원격 근무 규정' 로 바꿉니다
11  [retrieve] 질문='원격 근무 규정' -> 문서 0건
12  [rewrite]  2번째 재시도 -> 질문을 '재택근무 규정' 로 바꿉니다
13  [retrieve] 질문='재택근무 규정' -> 문서 1건
14  [generate] 문서로 답변을 만듭니다
15  지나간 경로: START -> retrieve -> rewrite -> retrieve -> rewrite -> retrieve -> generate -> END
16  최종 답변: 답변: 재택근무는 주 2회까지 가능하다.
17============================================================
18[끝내 못 찾음] 질문='집에서 일해도 되나요?', 후보=['원격 근무 규정', '홈오피스 규정']
19  [retrieve] 질문='집에서 일해도 되나요?' -> 문서 0건
20  [rewrite]  1번째 재시도 -> 질문을 '원격 근무 규정' 로 바꿉니다
21  [retrieve] 질문='원격 근무 규정' -> 문서 0건
22  [rewrite]  2번째 재시도 -> 질문을 '홈오피스 규정' 로 바꿉니다
23  [retrieve] 질문='홈오피스 규정' -> 문서 0건
24  [give_up]  재시도를 다 써서 포기합니다
25  지나간 경로: START -> retrieve -> rewrite -> retrieve -> rewrite -> retrieve -> give_up -> END
26  최종 답변: 죄송합니다. 관련 규정을 찾지 못했습니다.

22# ---------- 1) 도구 정의 ----------
23@tool
24def search_employee(name: str) -> str:
25    """임직원 이름으로 소속 팀과 이메일을 조회합니다. 임직원 정보 질문에 사용하세요."""
26    # 교육용 더미 데이터입니다. 실제 사내 데이터와 무관합니다.
27    employees = {
28        "김하늘": "김하늘 / 클라우드운영팀 / [email protected]",
29        "박도윤": "박도윤 / 물류플랫폼팀 / [email protected]",
30    }
31    return employees.get(name, f"'{name}' 님을 찾을 수 없습니다.")
32
33
34@tool
35def calculate(expression: str) -> str:
36    """수식 문자열을 계산해 정확한 결과를 반환합니다. 예: '34 * 27', '(15 - 5) / 2'"""
37    try:
38        return str(eval(expression, {"__builtins__": {}}, {}))
39    except Exception as e:
40        return f"계산 실패: {e}"
41
42
43tools = [search_employee, calculate]
44tools_by_name = {t.name: t for t in tools}   # 이름으로 도구를 찾기 위한 사전
45
46# ---------- 2) 도구 목록을 모델에 알려주기 ----------
47llm_with_tools = llm.bind_tools(tools)
48
49# ---------- 3) 대화 기록 초기화 ----------
50messages = [
51    SystemMessage(content="너는 삼성SDS 사내 어시스턴트야. 필요하면 도구를 사용해."),
52    HumanMessage(content="김하늘 님과 박도윤 님이 각각 어느 팀 소속인지 알려주고, 34 곱하기 27도 계산해줘."),
53]

55# ---------- 4) ReAct 루프: 도구 선택 -> 실행 -> 결과 관찰 -> 재판단 ----------
56MAX_TURNS = 10                                     # 안전장치
57turn = 0
58while turn < MAX_TURNS:
59    turn += 1
60    ai_message = llm_with_tools.invoke(messages)   # (추론) 모델이 도구 필요 여부를 판단
61    messages.append(ai_message)                    # 응답을 기록에 추가
62
63    if not ai_message.tool_calls:                  # 도구 요청이 없으면 최종 답변
64        print(f"\n[최종 답변] {get_text(ai_message)}")
65        break
66
67    for tool_call in ai_message.tool_calls:        # (행동) 요청된 도구를 전부 실행
68        tool_fn = tools_by_name[tool_call["name"]]
69        result = tool_fn.invoke(tool_call["args"])
70        print(f"[턴 {turn}] {tool_call['name']}({tool_call['args']}) -> {result}")
71
72        messages.append(                           # (관찰) 결과를 기록에 추가
73            ToolMessage(content=str(result), tool_call_id=tool_call["id"])
74        )
75    # while 처음으로 돌아감 -> 모델이 도구 결과를 보고 다시 판단 (재판단)
76else:
77    print("\n[중단] 최대 턴 수를 초과했습니다.")

1[턴 1] search_employee({'name': '김하늘'}) -> 김하늘 / 클라우드운영팀 / [email protected]
2[턴 1] search_employee({'name': '박도윤'}) -> 박도윤 / 물류플랫폼팀 / [email protected]
3[턴 1] calculate({'expression': '34 * 27'}) -> 918
4
5[최종 답변] ...

26@tool
27def search_employee(name: str) -> str:
28    """임직원 이름으로 소속 팀과 이메일을 조회합니다. 임직원 정보 질문에 사용하세요."""
29    # 교육용 더미 데이터입니다. 실제 사내 데이터와 무관합니다.
30    employees = {
31        "김하늘": {"team": "클라우드운영팀", "email": "[email protected]"},
32        "박도윤": {"team": "물류플랫폼팀", "email": "[email protected]"},
33    }
34    if name in employees:
35        e = employees[name]
36        return f"{name} / {e['team']} / {e['email']}"
37    return f"'{name}' 님을 찾을 수 없습니다. 조회 가능한 임직원: {list(employees.keys())}"
38
39
40@tool
41def calculate(expression: str) -> str:
42    """수식 문자열을 계산해 정확한 결과를 반환합니다. 예: '3480000 * 1.1', '48 * 1.12'"""
43    try:
44        return str(eval(expression, {"__builtins__": {}}, {}))
45    except Exception as e:
46        return f"계산 실패: {e}. 수식 형식을 확인하세요."
47
48
49@tool
50def classify_request(text: str) -> str:
51    """사내 문의 문장을 받아 담당 부서를 IT지원, 인사, 총무 중 하나로 분류합니다. 문의 접수와 부서 배정 질문에 사용하세요."""
52    # 도구 안에서 다시 LLM을 호출합니다. 도구 안의 AI입니다.
53    result = llm.invoke(
54        "다음 사내 문의의 담당 부서를 IT지원, 인사, 총무 중 하나로만 답하세요. "
55        f"다른 말은 붙이지 마세요.\n\n문의: {text}"
56    )
57    return get_text(result).strip()
58
59
60tools = [search_employee, calculate, classify_request]

64# [수동 루프의 messages 리스트] -> State + add_messages reducer
65class AgentState(TypedDict):
66    messages: Annotated[list, add_messages]
67
68
69SYSTEM_PROMPT = """너는 삼성SDS 사내 AI 어시스턴트야.
70
71[도구 사용 전략]
721. 임직원 정보는 반드시 search_employee로 조회해. 지어내지 마.
732. 계산이 필요하면 calculate를 사용해. 암산하지 마.
743. 사내 문의의 담당 부서 판단은 classify_request를 사용해.
754. 도구가 필요 없는 인사나 일반 질문에는 도구 없이 바로 답해."""
76
77llm_with_tools = llm.bind_tools(tools)
78
79
80# [수동 루프의 invoke와 append] -> agent 노드
81def agent(state: AgentState) -> dict:
82    messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
83    return {"messages": [llm_with_tools.invoke(messages)]}

 86# [수동 루프의 if not ai_message.tool_calls] -> 라우팅 함수
 87def should_continue(state: AgentState) -> Literal["tools", "end"]:
 88    last = state["messages"][-1]
 89    return "tools" if last.tool_calls else "end"
 90
 91
 92builder = StateGraph(AgentState)
 93builder.add_node("agent", agent)
 94builder.add_node("tools", ToolNode(tools))   # [도구 실행과 ToolMessage 포장] -> 내장 ToolNode
 95
 96builder.add_edge(START, "agent")
 97builder.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
 98builder.add_edge("tools", "agent")           # [while 루프] -> Cycle
 99
100graph = builder.compile()