Digital Nomad and FIRE

Flutter 앱을 개발하면서 가장 자주 사용 되는 UI 요소 중 하나가 바로 스크롤 가능한 리스트, ListView 입니다.

📌 ListView란?

ListView는 Flutter에서 수직 혹은 수평으로 스크롤 가능한 리스트 뷰를 제공하는 위젯 입니다. 빌더 패턴을 활용해 동적으로 아이템을 생성할 수 있습니다.

🧱 ListView의 주요 생성자

1. ListView()

• 고정된 children 리스트를 직접 전달.
  • 단점: 많은 항목이 있으면 성능 저하 발생 (모든 위젯을 메모리에 생성)

ListView(
  children: [
    Text('Item 1'),
    Text('Item 2'),
    Text('Item 3'),
  ],
)

(결과)

2. ListView.builder()

• 스크롤할 때마다 위젯을 생성하는 방식으로 성능에 유리.
• 항목이 많거나 동적으로 생성되는 경우에 필수.

ListView.builder(
  itemCount: 100,
  itemBuilder: (context, index) {
    return ListTile(title: Text('Item $index'));
  },
)

(결과)

3. ListView.separated()

• 각 항목 사이에 구분선이나 위젯을 넣고 싶을 때 사용.

ListView.separated(
  itemCount: 10,
  itemBuilder: (context, index) => ListTile(title: Text('Item $index')),
  separatorBuilder: (context, index) => Divider(), // 구분선
)

(결과)

4. ListView.custom()

• 고급 커스터마이징이 필요한 경우 사용.
• SliverChildDelegate를 통해 직접 동작 방식 정의 가능.

📏 주요 속성

🧠 주의할 점 & 팁

✅ 성능 최적화

• 많은 항목이 있을 경우에는 반드시 ListView.builder() 사용.
• shrinkWrap: true는 필요한 경우에만, 일반 상황에선 성능 저하 가능 (예: 다른 스크롤 뷰 안에 ListView 포함할 때).
• physics: NeverScrollableScrollPhysics() 로 스크롤 비활성화 가능.

✅ 자동 스크롤 위치 유지

• controller를 사용해 스크롤 위치 제어 가능.
  • 예: ScrollController, jumpTo(), animateTo() 등 활용.

📌 Scaffold 클래스란?

Scaffold는 Flutter에서 화면을 구성할 때 기본적인 뼈대를 제공하는 위젯(Widget)입니다. Material Design 앱의 기본 구조를 쉽게 구성할 수 있도록 도와줍니다.

📦 Scaffold 주요 속성

• appBar: 상단에 위치하는 앱 바
• drawer: 왼쪽 슬라이드 메뉴
• bottomNavigationBar: 하단 내비게이션 바
• floatingActionButton: 플로팅 액션 버튼
• body: 화면의 메인 콘텐츠 영역
• resizeToAvoidBottomInset : 키보드가 뜰 때 body가 자동으로 리사이즈될지 여부
• backgroundColor : 배경색 지정

💡 예제 확인

샘플로 로컬 프로젝트를 생성 하려면:

flutter create --sample=material.Scaffold.1 mysample

( 코드 내용 )

import 'package:flutter/material.dart';

/// Flutter code sample for [Scaffold].

void main() => runApp(const ScaffoldExampleApp());

class ScaffoldExampleApp extends StatelessWidget {
  const ScaffoldExampleApp({super.key});

  @override
  Widget build(BuildContext context) {
    return const MaterialApp(home: ScaffoldExample());
  }
}

class ScaffoldExample extends StatefulWidget {
  const ScaffoldExample({super.key});

  @override
  State<ScaffoldExample> createState() => _ScaffoldExampleState();
}

class _ScaffoldExampleState extends State<ScaffoldExample> {
  int _count = 0;

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('Sample Code')),
      body: Center(child: Text('You have pressed the button $_count times.')),
      floatingActionButton: FloatingActionButton(
        onPressed: () => setState(() => _count++),
        tooltip: 'Increment Counter',
        child: const Icon(Icons.add),
      ),
    );
  }
}

이 예제는 appBar 와 body로 구성되어 FloatingActionButton이 있는 Scaffold를 보여줍니다. FloatingActionButton 은 클릭 시 카운터를 증가시키도록 콜백 처리 됩니다.

결론

Flutter를 공부하는 입장에서 Scaffold는 매 화면을 구성할 때 출발점이 되는 위젯이니, 완벽하게 이해하고 넘어가는 것이 좋습니다.

Scaffold 없이 CustomScrollView 같은 걸로 직접 구성도 가능하지만, 처음에는 Scaffold를 적극 활용하는 것이 개발 생산성도 좋고 UI 구현도 훨씬 빠릅니다.

Reference

• Scaffold-Class

🧩 Flutter UI의 핵심 철학 : 모든 것이 위젯이다!

Flutter를 처음 접하면 가장 먼저 듣게 되는 말이 있습니다.

“ Flutter는 모든 것이 위젯이다! ”

Flutter에서 UI를 구성하는 모든 요소는 위젯(Widget) 입니다.

그 예시를 볼까요?

심지어 Padding, Margin, 애니메이션도 전부 위젯입니다!

🧱 위젯의 종류: Stateless vs Stateful

• StatelessWidget : 값이 변하지 않는 정적인 화면 요소
• StatefulWidget : 입력에 따라 바뀌는 동적인 화면 요소

화면에 입력창이나 버튼처럼 상호작용이 필요하다면 StatefulWidget이 쓰입니다.

🎯 UI는 어떻게 구성할까?

Flutter는 XML이나 Storyboard 없이, 코드로 UI를 직접 구성합니다.

즉, 위젯들을 조합하고 중첩하여 트리 구조(Tree) 를 만드는 방식입니다.

💡 예제 코드

Scaffold(
  appBar: AppBar(title: Text('Hello')),
  body: Center(
    child: Column(
      mainAxisAlignment: MainAxisAlignment.center,
      children: [
        Text('Welcome!'),
        ElevatedButton(
          onPressed: () {
            print('Clicked!');
          },
          child: Text('Click'),
        ),
      ],
    ),
  ),
);

• Scaffold는 페이지 전체 레이아웃을 담당
  • Scaffold는 Material Design의 시각적 Layout 구조를 표현함.
• AppBar, Text, Button 모두 위젯
• Center, Column은 레이아웃을 배치하는 위젯

이런 식으로 위젯 안에 또 다른 위젯이 들어가는 구조로 UI가 만들어집니다.

1. Cursor AI

Cursor AI는 AI 기반의 코드 편집기로, 개발자가 코드를 보다 빠르고 효율적으로 작성할 수 있도록 도와주는 도구입니다. 기존의 GitHub Copilot과 같은 AI 보조 도구와 유사하며, 강력한 코드 자동 완성 기능과 코드 리뷰 및 리팩토링 기능을 제공합니다.

Cursor AI의 주요 기능

• AI 기반 코드 자동 완성: 코드의 문맥을 이해하고 적절한 제안을 제공

• 코드 리뷰 및 리팩토링: 기존 코드의 품질을 높이도록 도와줌

• 내장형 터미널 및 IDE 기능: VS Code와 유사한 환경에서 동작

• 대화형 AI 개발 지원: ChatGPT를 활용한 코드 설명 및 개선 제안

2. MacBook에 Cursor AI 설치하기

2.1. Homebrew를 사용한 설치 (권장)

Homebrew를 사용하면 간편하게 Cursor AI를 설치할 수 있습니다.

brew install --cask cursor

설치가 완료되면 cursor 명령어를 실행하여 정상적으로 동작하는지 확인합니다.

cursor --version

2.2. 공식 웹사이트에서 설치

Homebrew를 사용하지 않는다면 Cursor AI 공식 웹사이트에서 설치 파일을 다운로드할 수 있습니다.

1. 웹사이트 접속 후 Mac용 설치 파일 다운로드

2. .dmg 파일을 실행하여 응용 프로그램 폴더에 드래그

   

3. 실행 후 초기 설정 진행

Cursor AI 활용하기

Cursor AI를 실행하고, 주로 작성하는 언어는 영어보다는 한국어를 사용이 많을 것임으로 한국어를 설정합니다.
터미널에서 Cursor AI를 사용하고자 하는 경우 Installed "cursor" 선택하여 설치합니다.

VS Code를 사용중이라면 USE Extensions 버튼을 클릭 합니다. 버튼을 클릭하면, VS Code에 설치된 Extension을 그대로 Cursor AI에 적용하게 됩니다.

데이터 보호 설정은 Privacy Mode로 하고, Continue 를 클릭합니다.

Curosor AI 계정이 있으면 그대로 로그인을 진행하고, 계정이 없는 경우 회원가입을 합니다.

소스코드를 관리하는 루트 디렉토리를 Open project 로 해당 폴더를 오픈하면, 하위 디렉토리 모두 불러와서 작업을 할 수 있습니다.

Cursor AI를 활용하여 개발 속도를 향상시키고, 코드 품질을 개선하는 데 도움이 되길 바랍니다!

Time Travel

Time Travel 기능은 대화형 에이전트 개발 시 과거의 대화 상태로 되돌아가거나 대화 경로를 재탐색할 수 있는 도구로 에이전트의 의사 결정 과정을 분석하고, 다양한 시나리오를 테스트하며, 대화 흐름을 최적화할 수 있습니다.

Time Travel 기능의 주요 이점:

• 대화 흐름 분석: 과거의 대화 상태로 돌아가 에이전트의 반응과 의사 결정 과정을 상세히 검토할 수 있습니다.

• 시나리오 테스트: 다양한 대화 경로를 탐색하여 에이전트의 반응을 테스트하고 개선할 수 있습니다.

• 버그 수정 및 최적화: 특정 시점으로 되돌아가 문제를 재현하고 수정하여 에이전트의 성능을 향상시킬 수 있습니다.

Implementation

이전 코드는 질의가 발생하면, human_assistance 툴(Tool)에서 interrupt가 발생되도록 되어 있으므로, 해당 툴을 거치지 않도록 툴 등록을 해제 합니다.

# tools.py


# 도구 목록
tools = [TavilySearchResults(max_results=2)]

1. 그래프(Graph) 실행

get_state_history() 메서드를 사용하여 특정 스레드의 상태 이력을 가져올 수 있습니다. 스냅샷을 통해 원하는 시점으로 되돌아가거나 대화 경로를 재 탐색할 수 있습니다.

다음은 두개의 질의를 전달 한 후, 특정 영역의 state를 다시 실행할 수 있도록 지정합니다. (여기서는 6번째 tool 호출 결과 내역)

# agent.py

# 그래프 실행
if __name__ == "__main__":

    config = {"configurable": {"thread_id": "USER_06"}}
    user_input = {"type": "user", "content": ("My Name is K. I'm learning LangGraph."
                  "Could you do some research on it for me?")}

    for event in graph.stream({"messages": [user_input]}, config, stream_mode="values"):
        event["messages"][-1].pretty_print()


    user_input = {"type": "user", "content": ("I would like to build chatbot agents using LangGraph.")}

    for event in graph.stream({"messages": [user_input]}, config, stream_mode="values"):
        event["messages"][-1].pretty_print()


    to_replay = None
    for state in graph.get_state_history(config):
        print("Num Messages: ", len(state.values["messages"]), "Next: ", state.next)
        print("-" * 80)
        if len(state.values["messages"]) == 6:
            # We are somewhat arbitrarily selecting a specific state based on the number of chat messages in the state.
            to_replay = state

    print(to_replay.next)
    print(to_replay.config)

(결과)

# ~~ 생략 ~~

Num Messages:  8 Next:  ()
--------------------------------------------------------------------------------
Num Messages:  7 Next:  ('chatbot',)
--------------------------------------------------------------------------------
Num Messages:  6 Next:  ('tools',)
--------------------------------------------------------------------------------
Num Messages:  5 Next:  ('chatbot',)
--------------------------------------------------------------------------------
Num Messages:  4 Next:  ('__start__',)
--------------------------------------------------------------------------------
Num Messages:  4 Next:  ()
--------------------------------------------------------------------------------
Num Messages:  3 Next:  ('chatbot',)
--------------------------------------------------------------------------------
Num Messages:  2 Next:  ('tools',)
--------------------------------------------------------------------------------
Num Messages:  1 Next:  ('chatbot',)
--------------------------------------------------------------------------------
Num Messages:  0 Next:  ('__start__',)
--------------------------------------------------------------------------------

{'configurable': {'thread_id': 'USER_06', 'checkpoint_ns': '', 'checkpoint_id': '1efe9bc1-fa9d-66aa-8006-85f90316985e'}}

메시지의 내용은 생략하고, 이후 결과들을 확인하면 6번째 호출은 도구를 호출 한 결과로써 to_replay.config 에는 checkpoint_id가 존재 하고 해당 id를 통해서 LangGraph의 체크포인터가 해당 시점의 state를 다시 가져올 수 있습니다.

checkpoint_id로 langsmith에서 확인할 수 있습니다.

Reference

• Time-Travel

Customizing State

LangGraph는 모델과 툴의 상태를 State로 정의하고, 이를 통해 상태의 변화를 추적합니다.
State의 변화 관리를 통해서 프로젝트의 요구 사항에 맞춰 상태를 효율적으로 관리하고 추적할 수 있게 해줍니다. 이를 통해 다양한 노드와 툴(Tool)이 상호작용하는 과정에서 발생할 수 있는 문제를 예방하고, 더 나은 사용자 경험을 제공할 수 있습니다.

Implementation

여기서는 이전 Agent의 내용을 바탕으로 챗봇이 검색 도구를 사용하여 특정 정보를 찾고 검토를 위해 사람에게 전달하는 시나리오를 구현합니다.

1. State에 키 추가

특정 정보를 찾아 State의 수정하기 위해서, 필요한 정보를 용이하게 접근할 수 있도록 state에 key 값을 추가합니다. 여기서는 name 과 birthday 키 값을 추가합니다.

# state.py


from typing import TypedDict, Annotated, Sequence
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    name: str
    birthday: str

2. State 업데이트

LangGraph에서 Human-in-the-loop 상호작용을 위한 human_assistance 도구를 다음과 같이 개선합니다.

✅ Call ID 기반의 State 업데이트

• Call ID를 활용하여 업데이트할 정확한 State를 찾도록 수정
• 각 실행 인스턴스를 구분하고 올바른 State 업데이트 보장

✅ 응답 정확도 판별 및 반환 값 변경

• y로 시작하는 응답("y")이면 결과를 반환, 아닌 경우에는 수정된 답변으로 반환
• 응답에서 "correct" 키가 없을 경우 기본값 빈 문자열 ("")을 반환

✅ InjectedToolCallId 사용하여 tool_call_id 관리

• LangChain 내부에서 tool_call_id를 자동 관리하도록 변경
• 각 툴의 실행 결과가 섞이지 않도록 분리하여 후속 작업의 혼선 방지

✅ Command를 활용한 State 업데이트

• State 변경 사항을 명확하게 추적 가능하도록 Command 사용

# tools.py

# 도구 정의
@tool
def human_assistance(name: str, birthday: str, tool_call_id: Annotated[str, InjectedToolCallId]) -> str:
    """Request assistance from a human."""
    human_response = interrupt({
        "question": "Is this correct?",
        "name": name,
        "birthday": birthday,
    })

    if human_response.get("correct", "").lower().startswith("y"):
        verified_name = name
        verified_birthday = birthday
        response = "Correct"
    else:
        verified_name = human_response.get("name", name)
        verified_birthday = human_response.get("birthday", birthday)
        response = f"Made a correction: {human_response}"

    state_update = {
        "name": verified_name,
        "birthday": verified_birthday,
        "messages": [ToolMessage(response, tool_call_id=tool_call_id)]  ,
    }

    return Command(update=state_update) # 상태 업데이트 명령어 반환

3. 그래프(Graph) 실행

상태(State) 업데이트 결과를 실행하여 확인합니다.
최근에 화두가 되었던, DeepSeek에 대한 출시일정을 확인하는 프롬프트(Prompt)를 구성해서 확인 해보겠습니다.

# agent.py

# 그래프 실행
if __name__ == "__main__":

    config = {"configurable": {"thread_id": "USER_03"}}
    user_input = {"type": "user", "content": "Can you look up when DeepSeek was released? \
When you have the answer, use the human_assistance tool for review."}

    for event in graph.stream({"messages": [user_input]}, config, stream_mode="values"):
        event["messages"][-1].pretty_print()

    human_command = Command(
        resume={
            "name": "DeepSeek-R1",
            "birthday": "Jan 20, 2025"
        }
    )

    for event in graph.stream(human_command, config, stream_mode="values"):
        event["messages"][-1].pretty_print()

(결과)

================================ Human Message =================================

Can you look up when DeepSeek was released? When you have the answer, use the human_assistance tool for review.
================================== Ai Message ==================================
Tool Calls:
  tavily_search_results_json (call_JK559rFs2TUTnOBXzq397glS)
 Call ID: call_JK559rFs2TUTnOBXzq397glS
  Args:
    query: DeepSeek release date
================================= Tool Message =================================
Name: tavily_search_results_json

[{"url": "https://en.wikipedia.org/wiki/DeepSeek", "content": "It would not be used for stock trading and would be
separate from High-Flyer's financial business.[4] In May 2023, the company was launched as DeepSeek.[2] DeepSeek's 
development is funded by High-Flyer.[3] After releasing DeepSeek-V2 in May 2024 which offered strong performance for 
a low price, DeepSeek became known as the catalyst for China's AI model price war. On 29 November 2023, DeepSeek 
launched DeepSeek LLM (large language model) which scaled up to 67B parameters. In December 2024, DeepSeek-V3 was 
launched. Benchmark tests showed it outperformed Llama 3.1 and Qwen 2.5 while matching GPT-4o and Claude 3.5 Sonnet.
[10][11] DeepSeek's optimization on limited resources highlighted potential limits of US sanctions on China's AI 
development.[12] \"Meet DeepSeek Chat, China's latest ChatGPT rival with a 67B model\". \"Chinese start-up DeepSeek's
new AI model outperforms Meta, OpenAI products\"."}, {"url": "https://www.techtarget.com/whatis/feature/DeepSeek-
explained-Everything-you-need-to-know", "content": "DeepSeek, a Chinese AI firm, is disrupting the industry with its 
low-cost, open source large language models, challenging U.S. tech giants. What is DeepSeek? DeepSeek focuses on 
developing open source LLMs. The company's first model was released in November 2023. DeepSeek Since the company was 
created in 2023, DeepSeek has released a series of generative AI models. The low-cost development threatens 
the business model of U.S. tech companies that have invested billions in AI. In contrast with OpenAI, which is 
proprietary technology, DeepSeek is open source and free, challenging the revenue model of U.S. companies charging 
monthly fees for AI services. Being based in China, DeepSeek challenges U.S. technological dominance in AI."}]
================================== Ai Message ==================================
Tool Calls:
  human_assistance (call_IBFdno2pWzb8gSFgFKfO7ySw)
 Call ID: call_IBFdno2pWzb8gSFgFKfO7ySw
  Args:
    name: DeepSeek
    birthday: November 2023
================================== Ai Message ==================================
Tool Calls:
  human_assistance (call_IBFdno2pWzb8gSFgFKfO7ySw)
 Call ID: call_IBFdno2pWzb8gSFgFKfO7ySw
  Args:
    name: DeepSeek
    birthday: November 2023
================================= Tool Message =================================
Name: human_assistance

Made a correction: {'name': 'DeepSeek-R1', 'birthday': 'Jan 20, 2025'}
================================== Ai Message ==================================

DeepSeek was launched in May 2023, with its first model released in November 2023. However, there was a correction 
indicating that the specific model "DeepSeek-R1" will be released on January 20, 2025.

결과를 확인하면, 검색 툴(tavily)을 통해 얻은 birthday는 November 2023 으로 human_assistance 툴에 interrupt 가 발생하였고, 리뷰를 통해 정정된 메시지로 전달 된 것을 확인 할 수 있다.
출시일을 birthday에 매핑되어 출력되어 의아하게 생각할 수 있는데, 이는 흔히 birthday 를 생일 뜻으로만 인식할 수 있는데 사전을 검색해보면 다른 뜻도 있음을 확인 할 수 있습니다.(부족한 영어실력이..발목을..😭😭)

3. 수동으로 State 업데이트 하기

graph.update_state를 활용하여 수동으로 thread의 키값으로 기반으로 state를 업데이트 할 수 있습니다.
기존 name 필드에 적용되어 있는 값을 DeepSeek-R1 Model로 이름을 변경하겠습니다.

# agent.py

# 그래프 실행
if __name__ == "__main__":

    print("================================= Graph State =================================")
    snapshot = graph.get_state(config)
    result = {k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}
    print(result)

    print("============================ Manually Updating State ==========================")
    update_state = graph.update_state(config, {"name": "DeepSeek-R1 Model"})
    print(update_state)    
    snapshot = graph.get_state(config)

    result = {k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}
    print(result)

(결과)

================================= Graph State =================================
{'name': 'DeepSeek-R1', 'birthday': 'Jan 20, 2025'}
============================ Manually Updating State ==========================
{'configurable': {'thread_id': 'USER_03', 'checkpoint_ns': '', 'checkpoint_id': '1efe8419-d950-6098-8006-3d30fc5fee6b'}}
{'name': 'DeepSeek-R1 Model', 'birthday': 'Jan 20, 2025'}

일반적으로 interrupt 기능을 사용하는 것이 권장 되며, 이를 통해 Human-in-the-loop 에서 필요한 정보만 전달하고, 불필요한 상태 변경을 방지할 수 있습니다.

4. LangGraph Studio에서 확인하기

LangGraph Studio에서 보다 편리하게 테스트 해볼 수 있습니다.

동일 질문의 질의하면, 검색 툴을 통해서 결과 반환된 후 Interrupt가 발생하고, Name과 Birthday를 입력할 수 있도록 커서가 이동 하였습니다.

업데이트 할 메시지를 입력하고, Resume을 클릭하며 내용의 변경사항과 최종 chatbot이 반환하는 메시지를 확인할 수 있습니다.

Reference

• Customizing State

AI 에이전트는 항상 예상대로 동작하지 않을 수 있으며, 일부 작업은 사람의 승인이 필요할 수도 있습니다. 이때, LangGraph의 Human-in-the-loop(HITL) 기능을 사용하면, 에이전트가 특정 작업을 완료하는 데 있어 사람의 입력을 받을 수 있게 되어 신뢰성과 제어를 향상시킬 수 있습니다.

Human-in-the-loop(HITL) 기능 개요

Human-in-the-loop(HITL)은 에이전트가 특정 작업을 수행하는 중간에 사용자의 승인을 받거나, 작업을 일시적으로 멈추고 사람의 피드백을 기다릴 수 있는 기능입니다. 이 기능을 통해 AI 시스템은 자동화된 작업 외에도 사람이 개입해야 하는 작업을 처리할 수 있으며, 결과적으로 더 높은 정확도와 안정성을 제공합니다.

LangGraph에서는 interrupt() 함수로 이러한 흐름을 제어할 수 있습니다. 이 함수는 작업 흐름을 일시적으로 멈추고, 사용자가 작업을 확인하거나 추가적인 입력을 제공할 수 있도록 합니다. 이후, 사용자가 입력을 제공하면, 작업이 다시 진행됩니다.

활용 시나리오

Human-in-the-loop(HITL) 기능은 여러 상황에서 유용하게 활용될 수 있으며, 이를 통해 자동화의 신뢰성을 높이고, 중요한 결정을 사람이 개입하여 더 정확하게 처리할 수 있습니다.

• 서비스 승인 및 배포: 자동화된 시스템이 서비스를 배포하기 전에, 관리자에게 승인을 요청하여 실수로 잘못된 배포가 이루어지지 않도록 합니다.

• 위험 관리: 중요한 결정을 내리기 전, 시스템이 자동으로 결정을 내리는 대신 사람의 의견을 반영하여 리스크를 최소화합니다.

• 고객 서비스: AI 챗봇이 고객과의 대화 중 민감한 정보를 처리할 때, 사람이 개입하여 확인 후 진행하는 방식으로 사용될 수 있습니다.

Implementation

LangGraph에서 Human-in-the-loop 기능을 활용하려면, 기존의 작업 흐름에서 특정 부분에 interrupt() 를 추가하면 됩니다. 이를 통해 에이전트가 중요한 결정을 내리기 전에 사람의 입력을 받을 수 있게 됩니다.

1. Human Assistant Tool 만들기

챗봇에서 사람의 개입을 할 수 있는 Tool을 정의하고, 도구목록에 human_assitance 함수를 추가합니다.

# tools.py

from langchain_community.tools.tavily_search import TavilySearchResults
from dotenv import load_dotenv
from langchain_core.tools import tool
from langgraph.types import interrupt

load_dotenv()

# 도구 정의
@tool
def human_assistance(query: str) -> str:
    """Request assistance from a human."""
    human_response = interrupt({"query": query})

    return human_response["data"]


# 도구 목록
tools = [TavilySearchResults(max_results=2), human_assistance]

2. 그래프 실행

#agent.py

# 그래프 실행
if __name__ == "__main__":

    config = {"configurable": {"thread_id": "USER_02"}}
    user_input = {"type": "user", "content": "AI 에이전트를 개발하기 위해 전문적인 가이드를 받고 싶다."}

    for event in graph.stream({"messages": [user_input]}, config, stream_mode="values"):
        event["messages"][-1].pretty_print()


    print("================================= Graph State =================================")
    snapshot = graph.get_state(config)
    print(snapshot.next)

    # "AI 에이전트를 개발하려면 안정적이고 확장이 용이한 LangGraph를 확인해보시기를 권장합니다."
    human_response = (        
        "If you want to develop AI agents, we recommend checking out LangGraph, which is reliable and scalable."
    )

    human_command = Command(resume={"data": human_response})

    # human_command = None

    events = graph.stream(human_command, config, stream_mode="values")
    for event in events:
        if "messages" in event:
            event["messages"][-1].pretty_print()

(결과)

================================ Human Message =================================

AI 에이전트를 개발하기 위해 전문적인 가이드를 받고 싶다.
================================== Ai Message ==================================
Tool Calls:
  human_assistance (call_Xyzp8czUPulqvGa9WsWT0qKf)
 Call ID: call_Xyzp8czUPulqvGa9WsWT0qKf
  Args:
    query: AI 에이전트 개발에 대한 전문적인 가이드 요청
================================= Graph State =================================
('tools',)
================================== Ai Message ==================================
Tool Calls:
  human_assistance (call_Xyzp8czUPulqvGa9WsWT0qKf)
 Call ID: call_Xyzp8czUPulqvGa9WsWT0qKf
  Args:
    query: AI 에이전트 개발에 대한 전문적인 가이드 요청
================================= Tool Message =================================
Name: human_assistance

If you want to develop AI agents, we recommend checking out LangGraph, which is reliable and scalable.
================================== Ai Message ==================================

AI 에이전트를 개발하기 위해 LangGraph를 추천합니다. 이 플랫폼은 신뢰할 수 있고 확장성이 뛰어나며, AI 에이전트를 구축하는 데 필요한 다양한 도구와 리소스를 제공합니다. LangGraph에 대한 자세한 정보와 사용 방법을 살펴보시면 좋겠습니다. 추가적인 도움이 필요하시면 말씀해 주세요!

• 결과를 확인하면, 사용자 질의(HumanMessage)가 발생한 후 human_assistance Tool에서 interrupt가 발생하여 그래프의 상태가 Tools에 머무른 것을 확인할 수 있습니다.
• 이후 그래프의 재개(Resume)를 하기 위해 Command의 인자값으로 resume 으로 전달하여 발생된 interrupt를 통과하여 그래프가 재개되었음을 확인할 수 있습니다.
• 최종적으로, 챗봇은 human_assistance의 Messsage를 통해서 최종적으로 결과를 전달하는 구조가 됨을 확인 할 수 있습니다.

Python의 내장 input()함수와 비슷하게, interrupt도구 내부에서 호출하면 실행이 일시 중지됩니다.
진행 상황은 체크포인터 선택에 따라 지속됩니다 .

4. LangSmith에서 추적 해보기

① Tool 호출단계에서 human_assitance에서 interrupt 가 발생한 부분을 확인할 수 있습니다.
② 그래프의 재개를 위해 Resume의 인자로 Data 키에 message가 들어간 부분을 확인할 수 있습니다.

5. LangGraph Studio에서 확인해보기

LagnGraph Studio에서 테스트를 수행해보면, 사용자 질의가 전달되면 Interrupt가 발생하여 답변이 멈춘상태를 확인할 수 있습니다.

Continue를 클릭하면 그래프 재개를 위한 Resume 박스가 만들어지고, 관련 메시지를 넣을 수 있습니다.

결과를 확인하면, human_assistance의 Tool에서 전달 메세지를 토대로 결과가 전달 되었음을 알 수 있습니다.

Reference

• Human-in-the-loop
• LangGraph-Types-Command
• 번역 LangGraph-Tutorial

LangGraph에서는 채크포인터(CheckPointer) 기능을 통해 챗봇의 상태를 저장하고 추적할 수 있습니다. 이를 활용하면 대화의 맥락을 유지하거나 에러 복구, Human-in-the-Loop(HITL), 그리고 Time-Travel 등 다양한 시나리오에서 유용하게 활용할 수 있습니다.

앞서 만든 Agent에서는 대화의 맥락을 이어갈 수 있는 형태의코드가 아니었습니다.

# agent.py

# ~~ 중략 ~~

# 그래프 빌더 초기화
graph_builder = StateGraph(AgentState, config_schema=GraphConfig)

# 'chatbot' 노드 추가
graph_builder.add_node("chatbot", chatbot)
# 'tool_node' 노드 추가
graph_builder.add_node("tools", tool_node)

# 'chatbot'을 엔트리포인트로 설정
graph_builder.set_entry_point("chatbot")

graph_builder.add_conditional_edges(
    "chatbot",
    route_tools,
    {"tools": "tools", END: END},
)

# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")

# 그래프 컴파일
graph = graph_builder.compile()

(결과)

================================ Human Message =================================

Hi there! My name is K.
================================== Ai Message ==================================

Hi K! How can I assist you today?
================================ Human Message =================================

Remember my name?
================================== Ai Message ==================================

I don't have the capability to remember personal information or previous interactions,
so I don't know your name. However, you can share it with me if you'd like!

Implementation

1. 체크포인터 만들기

MemorySaver를 사용하여 인메모리 체크포인터를 사용합니다.

# agent.py

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

• MemorySaver 객체 를 생성하고, checkpointer를 사용하여 그래프를 컴파일 및 상태를 저장합니다.

튜토리얼에서는 in-memory checkpointer 를 사용하지만,
프로덕션 단계에서는 SqliteSaver 또는 PostgresSaver 를 사용하여 자체 DB에 연결할 수 있습니다.

2. ToolNode와 tools_condition을 사용하여 그래프 구축하기

병렬 처리, 조건부 흐름 제어, 그래프 기반 동적 흐름 관리 등 시스템의 효율성, 확장성, 유연성을 향상시키고 다양한 동작을 효율적으로 처리할 수 있도록 하기 위해 prebuilt 된 ToolNode 와 tools_condition 으로 변경합니다.

• ToolNode : 그래프 상태(메시지 목록 포함)를 입력으로 받고 도구 호출의 결과로 상태 업데이트를 출력하는 LangChain Runnable
• tools_condition : state의 가장 최근 message를 가져와 tool_calls가 존재하는지 여부에 따라 분기

BasicToolNode → ToolNode로 변경

# nodes.py

import os, json
from langchain_openai import ChatOpenAI
from my_ai_agent.utils.state import AgentState
from my_ai_agent.utils.tools import tools
from dotenv import load_dotenv
from langchain_core.messages import ToolMessage
from langgraph.prebuilt import ToolNode, tools_condition

load_dotenv()

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key= os.getenv("OPENAI_API_KEY"),
    max_tokens=None,
    temperature=0.7,
)

# LLM에 도구를 바인딩
llm_with_tools = llm.bind_tools(tools)

system_prompt = "Chat with the AI assistant. You can ask questions about anything else."

# 'chatbot' 노드
def chatbot(state):
    """상태에서 메시지를 받아서 LLM을 호출하는 함수"""
    messages = state["messages"]
    messages = [{"role": "system", "content": system_prompt}] + messages
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}


# 도구 노드 생성    
tool_node = ToolNode(tools)

여기서부터는 gpt3.5로 하면 정확성이 너무 떨어집니다. 그나마 비용이 저렴한 40-mini로 바꿔서 진행합니다.

route_table → tools_condition 함수로 변경

# agent.py

import os
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from my_ai_agent.utils.state import AgentState
from my_ai_agent.utils.nodes import chatbot, tool_node
from langgraph.prebuilt import tools_condition


# Define the config
class GraphConfig(TypedDict):
    model_name: Literal["anthropic", "openai"]


# 그래프 빌더 초기화
graph_builder = StateGraph(AgentState, config_schema=GraphConfig)

# 'chatbot' 노드 추가
graph_builder.add_node("chatbot", chatbot)
# 'tool_node' 노드 추가
graph_builder.add_node("tools", tool_node)


# 'chatbot'을 엔트리포인트로 설정
graph_builder.set_entry_point("chatbot")


# 'chatbot'에서 'tool_node'로 이동하는 조건 추가
# The `tools_condition` function returns "tools" if the chatbot asks to use a tool, and "END" if
# it is fine directly responding. This conditional routing defines the main agent loop.
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
    # The following dictionary lets you tell the graph to interpret the condition's outputs as a specific node
    # It defaults to the identity function, but if you
    # want to use a node named something else apart from "tools",
    # You can update the value of the dictionary to something else
    # e.g., "tools": "my_tools"
    {"tools": "tools", END: END},
)

# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")


# 그래프 컴파일
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

LangGraph Cloud나 LangGraph Studio를 사용하는 경우, 그래프를 컴파일할 때 자동으로 checkpointer을 포함한 지속성 및 상태 관리를 처리하기 때문에 선언할 필요는 없습니다.

3. 그래프 실행

이전 메시지를 기억해서 답변 하는지 확인합니다.

# agent.py

# 그래프 실행
if __name__ == "__main__":

    config = {"configurable": {"thread_id": "USER_01"}}
    user_input = {"type": "user", "content": "Hi there! My name is K."}
    for event in graph.stream({"messages": [user_input]}, config, stream_mode="values"):
        event["messages"][-1].pretty_print()

    user_input = {"type": "user", "content": "Remember my name?"}
    for chunk in graph.stream({"messages": [user_input]}, config, stream_mode="values"):
        chunk["messages"][-1].pretty_print()

(결과)

================================ Human Message =================================

Hi there! My name is K.
================================== Ai Message ==================================

Hello, K! How can I assist you today?
================================ Human Message =================================

Remember my name?
================================== Ai Message ==================================

Yes, I remember your name is K. How can I assist you further?

4. Langgraph studio에서 확인해보기

LangGraph Studio를 열면 자동으로 새 스레드 창이 열립니다. 기존 스레드가 열려 있는 경우 다음 단계에 따라 새 스레드를 만듭니다.
우측 상단의 + 버튼을 클릭하여 새로운 스레드를 생성할 수 있습니다.

Reference

• Adding Memory to the chatbot
• 그래프에 영속성 추가하는 방법
• Migrating from ConversationalChain
• Prebuilt components

LangGraph 에서 Tool을 활용하여 어떻게 활용할 수 있는지에 대해 설명합니다. ChatBot이 답할 수 없는 질문을 처리하기 위해 웹 검색이 가능한 도구와 통합하여 관련정보를 찾고 더 나은 답변을 제공할 수 있도록 연계합니다.

Tool 사용하기

LangGraph에서 Tool은 함수나 API로 데이터를 처리하는 역할을 합니다. 예를 들어, 검색 기능을 구현하고 싶다면, 웹 검색 API나 데이터베이스 쿼리, 혹은 파일 시스템에서 데이터를 검색하는 도구를 활용할 수 있습니다.

Tool 설정하기

LangGraph에서 Tool을 설정하는 방법은 간단합니다. 먼저, Tool을 정의하고, 그것을 노드에서 사용할 수 있도록 연결해야 합니다.

환경설정

AI Agent 검색 엔진기반의 Tavily 검색엔진 패키지를 설치합니다.

poetry add tavily-python langchain_community

TAVILY API 키 설정

.env 에 TAVILY_API_KEY를 설정합니다.

TAVILY_API_KEY= "<your-api-key>"

Implementation

1. Tool 정의

검색 기능을 위한 Tool을 정의합니다.

# tools.py

from langchain_community.tools.tavily_search import TavilySearchResults
from dotenv import load_dotenv

load_dotenv()

tools = [TavilySearchResults(max_results=1)]

간단하게 Tavily 검색을 해볼 수 있습니다.

tools[0].invoke("what is Langchain?")

2. Tool 등록

LLM 모델에 bind_tools()를 사용하여 도구를 바인딩 합니다.
결과는 ToolCall Object 에 저장되고, .tool_calls 를 확인하여 도구 호출 결과를 확인할 수 있습니다.

# node.py

# LLM에 도구를 바인딩
llm_with_tools = llm.bind_tools(tools)

system_prompt = "Chat with the AI assistant. You can ask questions about anything else."

# 'chatbot' 노드
def chatbot(state):
    """상태에서 메시지를 받아서 LLM을 호출하는 함수"""
    messages = state["messages"]
    messages = [{"role": "system", "content": system_prompt}] + messages
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

3. Tool 노드(Node) 정의

가장 최근의 메시지에 tool_calls가 포함되어 있는 경우, 도구를 호출하는 노드를 구현합니다.

# node.py

from langchain_core.messages import ToolMessage

# 도구 노드 정의
class BasicToolNode:

    def __init__(self, tools: list) -> None:
        self.tool_by_name = {tool.name: tool for tool in tools}

    def __call__(self, inputs: dict):
        if messages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raise ValueError("No messages found in input")
        print(message)
        outputs = []
        for tool_call in message.tool_calls:
            tool_result = self.tool_by_name[tool_call["name"]].invoke(tool_call["args"])

            outputs.append(ToolMessage(
                content=json.dumps(tool_result),
                name=tool_call["name"],
                tool_call_id=tool_call["id"]
            ))

        return {"messages": outputs}

# 도구 노드 생성    
tool_node = BasicToolNode(tools)

4. 그래프(Graph) 정의

Tool 노드를 그래프에 추가 합니다.

# agent.py

from my_ai_agent.utils.node import tool_node

graph_builder.add_node("tools", tool_node)

도구 노드가 추가되면 conditional_edges를 정의할 수 있습니다.

5. 조건부엣지(Conditional edges) 정의

조건부 엣지(Conditional edges) 는 현재의 그래프의 상태에 따라 다른 노드로 라우팅을 할 수 있습니다.

• 조건부 엣지는 단일 노드에서 시작합니다. 즉, chatbot 노드가 실행될 때 다음 동작을 어떻게 처리할지 결정하는 역할을 합니다.
• 다음에 호출할 노드를 나타내는 문자열이나 문자열 목록을 반환함.

① 챗봇의 마지막 상태에서 tool_calls 가 있으면, tools 호출하는 함수를 생성합니다.
함수안에 도구 호출이 이루어지지 않으면 END 를 반환 하므로 finish_point 를 명시적으로 설정할 필요가 없습니다.

# agent.py

from langgraph.graph import END

def route_tools(state: AgentState):
    """
    Use in the conditional_edge to route to the ToolNode if the last message has tool calls. Otherwise, route to the end.
    """
    if isinstance(state, list):
        ai_messages = state[-1]
    elif messages := state.get("messages", []):
        ai_messages = messages[-1]
    else:
        raise ValueError(f"No messages found in input state to too_edge: {state}")

    if hasattr(ai_messages, "tool_calls") and len(ai_messages.tool_calls) > 0:  # Check if the last message has tool calls
        return "tools"

    return END

② 챗봇 노드가 완료될 때마다 이 함수(route_tools)를 통해 다음으로 어디로 가야 하는지 확인하도록 그래프에 선언합니다. (add_conditional_edges)

# agent.py

# 'chatbot'에서 'tool_node'로 이동하는 조건 추가
# The `tools_condition` function returns "tools" if the chatbot asks to use a tool, and "END" if
# it is fine directly responding. This conditional routing defines the main agent loop.
graph_builder.add_conditional_edges(
    "chatbot",
    route_tools,
    # The following dictionary lets you tell the graph to interpret the condition's outputs as a specific node
    # It defaults to the identity function, but if you
    # want to use a node named something else apart from "tools",
    # You can update the value of the dictionary to something else
    # e.g., "tools": "my_tools"
    {"tools": "tools", END: END},
)

③ 그리고, chatbot 노드에서 tool 이 호출되고 난 이후에 chatbot 노드가 다음 스텝을 결정할 수 있도록 경로를 설정합니다.

# agent.py

# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")

그래프 컴파일(Compile)

compile() 메서드는 정의된 그래프를 실행 가능한 형태로 변환합니다.

# agent.py

graph = graph_builder.compile()

그래프의 실행결과를 확인합니다. (LangStudio)

Reference

• Tavily Search
• Enhancing the Chatbot with Tools