260930

u·3일 전

TIL

목록 보기
28/30

사이클(self-correction 루프)

: 작성 → 검토 → 수정 → 검토

  1. State 정의

  2. Node 생성

  3. Edge 연결

    1. 문서 검토 통과
      1. 종료
    2. 미통과 & 검토 횟수 남음
      1. 수정
    3. 미통과 & 검토 상환 도달
      1. 종료 (사람이 확인)
    builder.add_edge("draft_resume", "review_resume")

# 조건부 엣지 -> 루프
builder.add_conditional_edges(
		"review_resume", choose_next, 
		{"revise": "revise_resume", "end": END,})
		
builder.add_edge("revise_resume", "review_resume")

체크포인터 checkpointer (단기 상태 영속화)

  1. State 정의

    1. prompt → MessagesPlaceholder 넣기

  2. Node 생성

  3. Edge 연결

    1. checkpointer 넣기

      # 여러 invoke가 같은 체크포인터 객체에 저장된 대화를 사용
checkpointer = InMemorySaver() # 메모리에 저장하는 체크포인터
chat_graph = chat_builder.compile(checkpointer=checkpointer) # <-- 체크포인트 넣어줌!

  4. 답변

    1. config
    # 첫 호출에는 지원 자료와 질문을 함께 저장
config_a = {"configurable": {"thread_id": "career-A"}}

question_a1 = "공고와 연결해 제 경험 중 이력서에 강조할 것을 하나 골라 주세요."
first_turn = chat_graph.invoke({
    "job_posting": job_posting,
    "company_info": company_info,
    "applicant_info": applicant_info,
    "messages": [HumanMessage(content=question_a1)],
}, config_a)

# 같은 thread_id에 새 질문만 전달하면 저장된 자료와 이전 대화를 이어서 읽음
question_a2 = "그 프로젝트에서 제가 담당한 업무가 드러나도록 이력서에 넣을 한 문장으로 써 주세요."
second_turn = chat_graph.invoke({"messages": [HumanMessage(content=question_a2)]}, config_a)
  • checkpointer : 갱신된 State + 실행 위치를 저장 , state 저장하는 데이터 저장소
  • add_messages : 새 ID의 메시지는 추가하고, 같은 ID의 메시지는 교체
  • thread_id : 저장된 여러 작업 중 어느 작업을 이어갈지 구분 , 특정 세션 ID - 사용자별 또는 대화별 특정 연관된 대화 그룹
    • 단, 멀티턴이 되려면
    1. State에 messages가 누적되어야 하고
    2. 프롬프트가 MessagesPlaceholder로 그 messages를 실제 LLM 입력에 넣어야 함
  • MessagesPlaceholder : State에 누적된 대화를 모델에 전달
  • InMemorySaver : 현재 프로세스 메모리에 저장. 같은 객체를 재사용해야 하며 커널이 종료되면 기록이 사라짐

멀티턴 대화랑 체크포인터 차이점

  • 싱글턴 & 멀티턴
    • 싱글턴 : 이전 대화를 기억 못 하고 현재 질문에 대한 응답만 함
    • 멀티턴 대화 : 사용자의 질문과 AI의 응답을 누적해서 LLM의 최근 응답시에 이전 대화들을 프롬프트에 입력으로 넣어서 답변을 생성
      • 체크포인터는 멀티턴 구현에 사용할 수 있는 상태 저장 수단
      • 체크포인터 : 상태를 저장, 상태에 대화 목록이 저장되어있어야하고
        LLM의 프롬프트에 대화 목록을 넣기 위한 messageplaceholder가 있어야 ← 멀티턴 대화가 가능

현재 대화 A가 끝났다

체크포인터에서 Thread_id 기반으로 최신 State를 가져와서
→ 다시 이어서 대화 가능 (무조건 XX)

  • 전제 조건
    • (1) 상태에 대화 목록이 저장되어있어야하고
    • (2) LLM의 프롬프트에 대화 목록을 넣기 위한 MessagesPlaceholder가 있어야 함

messages 누적 + 체크포인터 + 같은 thread_id + MessagesPlaceholder → 이전 대화를 이어가는 멀티턴 가능

사람검토 (Human In The Loop, HITL)

사용자에게 질문, 사용자 검토, 권한 확인 요청

  • 핵심 흐름
    interrupt (검토 요청, 초안과 선택지 → 호출자에게 전달) → 중단 → get_state → 현재 interrupt 확인 → Command(resume=...) → 같은 thread_id로 재개 → 재실행된 노드의 interrupt가 그 응답을 반환
  1. 그래프 실행
  2. interrupt()에서 멈춤
  3. get_state(config)로 현재 상태 조회
  4. snapshot.interrupts[0]로 현재 요청 확인
  5. 사람이 응답
  6. Command(resume=...)로 재개
  7. 같은 thread_id로 이어서 실행
  8. 또 interrupt()면 다시 멈춤
  9. 없으면 END
# 현재 작업 상태 묶음
snapshot = mail_graph.get_state(mail_config)

# 현재 state 데이터
snapshot.values["draft"]

# 현재 사람이 응답하길 기다리는 interrupt
snapshot.interrupts[0]

# 사람에게 요청을 보여주고 그래프를 멈춤
response = interrupt(request)

# 사람이 준 응답을 전달하면서 멈춘 그래프를 다시 실행
Command(resume={request.id: {"action": "approve"}})

# 수정 요청 전달
result = mail_graph.invoke(
    Command(
        resume={
            request.id: {
                "action": "edit",
                "feedback": "더 짧게 수정해주세요."
            }
        }
    ),
    mail_config
)

# 수정 후 다시 interrupt()에서 멈췄다면:
revised_snapshot = mail_graph.get_state(mail_config)
revised_request = revised_snapshot.interrupts[0]

  1. State 정의

  2. Node 생성

    1. interrupt : 멈추고 사람의 명령을 기다림
    def human_review(state: NoticeState):
    """초안을 전달해 대기하고, 사람의 선택과 수정 내용을 반환합니다."""
    print("검토 노드 시작")  # 재개할 때도 이 줄부터 다시 실행되는지 확인

    # graph의 실행이 멈추고, 사람의 입력을 받도록 대기 !!!
    decision = interrupt(
        # 아래의 딕셔너리는 정의된 형태의 딕셔너리 아님! 사용자에게 안내할 내용을 딕셔너리로 정리함 , 문자열 넣어도됨
        {
        "stage": "notice_review",  # 화면에서 안내문 검토 단계임을 구분할 이름
        "draft": state["draft"],
        "actions": ["approve", "edit", "reject"],
        "guide": "approve: 승인, edit: 수정 요청, reject: 반려",
    })  # 초안과 선택지를 전달하고 중단. 재개 시 사람의 응답 딕셔너리를 반환

    # command(resume) <-- 동작 재개
    # decision 딕셔너리 key : action, feedback, edited_text
    return {
        "decision": decision["action"],  # 응답 전체에서 선택한 경로만 State에 저장
        "feedback": decision.get("feedback", ""),
        # 사람이 최종 문장을 직접 고쳤다면 그 내용을 새 초안으로 반영
        "draft": decision.get("edited_text", state["draft"]),
    }

  3. Edge 연결

streaming

# 스트리밍 응답
# (1) 중간 진행 상황 확인 하기 위한 용도
# (2) 최종 응답에 대해서 스트리밍으로 받기 위한 용도 -> 글자가 쭈르르륵 (다 완성해서 받는게 아니라 생성할 때마다 받기)

v2 vs v3 스트리밍

  • v2 → stream(): 여러 결과가 섞여 나옴 → type으로 구분하고 data에서 값 확인
  • v3 → stream_events(): 실행 객체(run)로 반환 → 원하는 결과를 속성으로 바로 확인
    • run.output → 최종 결과

SQLite

: LangGraph의 State와 실행 위치를 SQLite 파일에 저장 → 커널을 재시작해도 작업을 이어서 실행

  • InMemorySaver & SqliteSaver
    • InMemorySaver : 메모리에 저장 → 커널 종료 시 사라짐
    • SqliteSaver : .sqlite 파일에 저장 → 커널 종료 후에도 유지
  • 같은 DB 파일 + thread_id → 이전 State 조회 및 이어가기
  1. SQLite 파일 준비
  2. SqliteSaver로 DB 연결
  3. compile(checkpointer=checkpointer)로 그래프에 연결
  4. thread_id를 지정해서 실행
  5. State와 실행 위치가 SQLite에 저장
  6. 재시작 후 같은 DB + thread_id로 복원
# SQLite 연결 + 그래프 생성
with SqliteSaver.from_conn_string(str(chat_db)) as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)

    # 저장된 State 조회
    saved = graph.get_state(config)

    # 실행 → State가 SQLite에 저장
    graph.invoke(input, config)
profile
u
이전 포스트

260929

다음 포스트

261001

0개의 댓글