DEV Community: Jaeyoun Nam

HAOS 입문기 네트워크 설정

Jaeyoun Nam — Wed, 30 Jul 2025 12:41:39 +0000

공유기에서 static 주소 할당

HAOS는 항상 같은 IP로 물려있는게 좋음.

DHCP 할당 정보 탭으로 가서
현재 homeassistant로 되어 있는 녀석의 Mac 주소를 알아낸 후
DHCP 고정할당으로 가서 '192.168.xxx.100'을 부여한다.

외부 접속

Port fowarding으로 뚫는 방법도 있지만, 그보다 설정이 더 쉬운 tunneling을 통해 외부와 연결해줄 것이다.

https://github.com/brenner-tobias/addon-cloudflared/wiki/How-tos

cloudflare의 tunnel을 사용할껀데, 위의 링크를 따라하면 된다.

AIDER가 코드베이스 인덱싱 하는 법

Jaeyoun Nam — Sat, 26 Apr 2025 08:19:34 +0000

Cursor같은 Code agent가 어떻게 코드 베이스를 검색하고 관련 코드를 읽어서 동작하는지 알고싶어서, 비슷한데 오픈소스인 aider를 둘러보다가 블로그 글을 찾아서 정리한다.

aider는 repository map이라는 것을 만들어서 AI Agent가 관련된 코드를 찾아서 활용할 수 있도록 한다.

블로그 링크

AI는 코드베이스 전부를 알고 있을 때, 어떤 테스크가 주어졌을 때 해결하는 걸 잘한다.
하지만 코드베이스가 커지면 테스크와 관련된 정보 중 누락하는게 발생하게되고, 그러면 성능이 떨어진다.
따라서 관련 정보를 최대한 잘 줄 수 있도록 AI를 위한 code search 툴을 매우 잘 제공할 필요가 있다.

Repo Map 이라는 걸 빌드해서 컨텍스트를 제공한다.

repo map에는 중요한 심볼들이 저장되어 있고, 이를 LLM에게 넘겨주는 방식으로 코드베이스 전반에 대한 지식을 넘겨준다.

aider/coders/base_coder.py:
    class Coder:
        abs_fnames = None
    ...
    @classmethod
    def create(self, main_model, edit_format, io, skip_model_availabily_check=False, **kwargs):
    ...
    def abs_root_path(self, path):
    ...
    def run(self, with_message=None):
    ...

aider/commands.py:
    class Commands:
        voice = None
    ...
    def get_commands(self):
    ...
    def get_command_completions(self, cmd_name, partial):
    ...
    def run(self, inp):
    ...

하지만 레포의 사이즈가 크다면 전부 넘겨주지는 못할 것이다. 그래서 repo map 에서도 관련이 있는 부분을 ranking하는 로직이 있어서, 관련성이 높은 부분만 추려서 보낸다.

좋은 Repo map 만들기

Aider는 Tree-sitter를 이용해서 Repo map을 만든다.

Tree sitter는 전체 코드베이스를 AST(Abstract Syntax Tree)로 만드는데, AST를 이용하면 우리는 함수, 클래스, 변수, 타입 등이 어디에 쓰이는 지 알 수 있다.

AI code search

결국 어떤 정보들을 잘 넣어주느냐가 아직도 중요하다. AI가 기존 코드베이스에서 잘 찾아서 일을 수행할 수 있게 하려면 좋은 AI code search 툴이 필요하다.

Why MCP

Jaeyoun Nam — Wed, 12 Mar 2025 11:43:50 +0000

MCP가 11월 25 (한국시간)에 발표되고 나서 이제 3달정도가 지났다.

MCP is Protocol

MCP는 Model Context Protocol의 약자로 Protocol이 함유하는 뜻 그대로 통신 규약이다.

규약이기 때문에 사실 그 자체로 대단한 기술은 아니다. 슈퍼파워네 뭐네 하는 소리가 있어도 그건 조금 과장된 것이다. 하지만 (슈퍼)파워를 AI Agent한테 부여하기 쉬워진것은 맞다.

규약이 유용하려면 프로토콜이 잘 설계 되야하는 것도 있지만 한가지 조건이 더 있다. 바로 유명해야 한다는 것이다.

우리가 인터넷 통신할때 쓰는 TCP/IP, RestAPI, 코딩할 때 쓰는 다양한 Interface, 조금 만 더 나가면 우리가 대화에 쓰는 언어까지 서로 내용을 공유하는 사람이 많으면 그 가치가 올라가는 것들이다.

거기서 오는 특성이 하나 더 있다. 점점 대체 불가능해진다는 것이다.

TCP/IP는 말해 뭐하고, OpenAI가 처음 GPT를 발표하면서 내놓은 API는 그 형태 그대로 다른 모델들에서도 사용해서 이제 OpenAI의 API 형식을 따르지 않으면 안된다.

Why MCP?

MCP는 LLM이 활용할 수 있는 Tool, Prompt, Resource(Data) 관리자와 소통할 수 있게 해주는 프로토콜이다. Tool, Prompt, Resource를 쉽게 AI가 가져다 쓸 수 있도록 돕는다고 생각하면 된다.

Tool의 개념은 GPT 서기 1년부터 존재해오던 개념이다. 사실 Agent의 개념과 거의 동시에 나왔다. Agent로써의 LLM은 두뇌의 역할이고, 인간이 도구를 사용해서 문제를 해결하듯 Agent가 Tool을 사용한다는 개념은 아주 직관적인 개념이다.

그런데 문제는 이 Tool을 모두가 다른 방식으로 비슷한 것을 만들고 있었다는 것이다. 아마 슬랙 연동하는 툴을 몇만명이 만들지 않았을까...

이런 문제의식을 가지고 이런 툴들을 묶어서 제공하려는 곳들도 있었고 Composio, GPTs action(못들어 봤을 수 있는데 OpenAI꺼다)은 Interface를 제공하고 이를 구현하면 GPT가 사용할 수 있도록 만들어주기도 했다. MCP 비슷하게 표준을 제안한 곳도 있다.AgentsJson

하지만 너무 닫혀있는 시스템이다거나 안유명하거나의 이유로 이들은 표준이 되지 못했다. MCP도 발표할때는 표준은 아니었지만 이제 표준에 가까워졌다.

Challange of MCP

지금 MCP의 상태는 초기 생태계 상태이다. MCP Server를 아무나 만들어서 공유할 수 있는데, 이 공유하는 생태계가 아직 부족하다.

MCP server는 기본적으로 남의 코드이다. 그래서 아무 MCP Server나 좋아보여서 돌렸다가는 거덜날 가능성이 있다. npm 같은 marketplace가 있고 보안 검증들도 철저하게 이루어져야한다.

대부분의 MCP Server는 퀄리티가 안좋다.

핫해지고 좀 알려지긴했지만 표준은 아니다.
MCP 등록을 해야하고 등록시 인증 과정이 불편하다. 내 Agent가 어떤 MCP Server의 툴을 쓰게 하고 싶으면 그 서버를 이용하기 위한 토큰을 포함하여 MCP서버를 등록해야한다.

이건 하면 되지 싶지만, 사실 완전한 Agent라면 필요한 툴을 찾아서 알아서 척척 내가 주문한 내용을 완수해야한다.

"깃헙의 내 레포의 가장 최신 이슈를 해결해서 PR을 올리고 슬랙으로 알림을 보내줘"라는 테스크를 주는 상황이다.

완전한 Agent: "깃헙을 써야겠네, 깃헙 MCP Server에서 정보를 가져와 볼까, 토큰이 필요하네, (뭐 어케해서 PAT얻어옴), 깃헙 MCP로 최신이슈 조회, ...., 어 슬랙도 필요하네, 슬랙 MCP Server를 찾아봐야지, ...., 전송 완료"

지금: 깃허브도 필요하고 슬렉도 필요하니까 MCP서버 설치링크를 찾아서 설치하고 인증 등록하고 "해줘"

Future of MCP

위의 문제들은 centralized registry 와 well-known 파일들로 해결을 하려고 하는 것으로 보인다.

Why Now?

지금 MCP를 도입해봐야하는 이유는 지금 꽤 괜찮은 걸 올리면 쉽게 유명해질 수 있다는 것이다.

Why Not API?

API도 사실 OpenAPI 같은 스펙이 있고, 사람도 충분히 이해할 수 있기에 당연히 LLM도 이해하고 사용할 수 있다.

1년전에는 AI가 웹을 돌아다닐 수 있기 때문에, 그냥 쓰고 싶은 API찾아서 쓰면 된다고 생각했다. 아니면 이런 퍼블릭 API 리스트도 학습해서 그냥 알고 있어서 쓰거나 하게 될 줄 알았다.

근데 지금와서 API를 툴로 활용하도록 발전한게 아니라 MCP라는 새로운 규약이 등장했다. 어떤 의미에서 MCP가 API보다 더 가치 있을까 생각해보았다.

API가 Agent의 비해 상대적으로 Low level이다. 우리가 Agent한테 명령하는 일들은 우리가 수 분을 소비해야하는 일들이다. 이 고수준의 일을 상대적으로 저수준인 API의 집합으로 나누려면 수십개의 API 호출을 해야할지도 모른다.

사람은 일을 할때 UI를 통해서 작업을 한다. UI에 있는 버튼 하나에 사실 많은 API들이 숨겨져 있는 거라고 볼 수 있다. Agent도 그런 버튼 같은게 필요하다.

API는 바뀔 확률이 높다. API가 바뀔때 마다 모든 Agent는 그 정보를 업데이트 해야한다.

반면 MCP는 추상화 레벨이 높아서 잘 바뀌지 않는다. 예를 들어서, "컴퓨터를 켠다" 라는 액션이 있다. 그 컴퓨터의 부품들의 키는 방식이 어떻게 바뀌든 유저는 알필요 없다. "컴퓨터를 키는 버튼"은 항상 그대로다. 컴퓨터를 조립하는 곳 한군데서만 신경 쓰면 된다.

디자인 패턴의 Facade패턴과 비슷하다.

Consistency Models and Replicache

Jaeyoun Nam — Wed, 01 Jan 2025 12:12:24 +0000

Consistency Model

Consistency Model은 분산 시스템에서 어느정도의 Consistency를 제공할 것이냐에 대한 여러가지 규약이다. 이전 포스트 중 PACELC 이론에서 Latency와 Consistency를 Trade off 하는 내용이 있는데, 이 Trade off 에 따라서 어느정도의 Consistency 를 제공할 것인가에 대한 내용이다.

이전 포스트에서 다룬 Replicache도 분산 시스템의 일종이다. Replicache는 Causal+ Consistency를 따른다고 한다.

출처

Strict Consistency

항상 시간 순서대로 값을 읽을 수 있는 Consistency이다. 값이 쓰여진 시점 이후에 발생하는 모든 읽기는 항상 그 값을 읽는다. 이를 만족 하려면 두가지의 조건이 필요하다. 모든 노드가 공유하는 global clock이 존재하고 이에 동기화 해야하며, Write가 즉시 수행되어야한다.

w(x)a: x에 a라는 값을 write, r(x)a: x에서 a란 값을 read

t시간에 w(x)a 이 1번 노드에서 발생하고, t+delta(0.0000000000000001s)후에 2번 노드에서 r(x)가 발생했을때, 2번노드는 값 a를 읽어야한다.

같은 t시간에 operation이 수행되지 않도록 global clock이 존재해서 operation을 직렬화해야하고, write가 즉시 수행되어서 t+delta 시간에 write한 값을 읽을 수 있어야한다.

하지만 이는 실제로 불가능한 일이다. 따라서 Strict Consistency는 이론으로 존재한다.

아래 상황에서 Strict Consistency를 준수한다면

정답은 아래와 같다.

Sequential Consistency

Sequential Consistency의 핵심 키워드는 'Global Ordering'이다. 모든 쓰기/읽기 연산은 하나의 통합된 순서가 있는 것 처럼 동작해야한다. 또 각각의 프로세스(클라이언트)가 수행하는 연산은 그 순서대로 실행되어야 한다.

아래의 경우 p3와 p4는 어떤 값을 읽을 수 있을까?

아래는 Strict Consistency와 같으므로 Sequential Consistency도 만족한다.

아래는 Global Ordering을 찾을 수 있으므로 Sequential Consistency가 맞다.

Global Ordering: w(x)a, r(x)a, w(x)b, r(x)b

하지만 아래의 경우에는 a가 b보다 먼저 쓰여졌는데 나중에 읽힌다. 이런 경우는 Sequential Consistency를 만족하지 않는다.

Causal Consistency

Causal 이란 원인-결과 관계를 의미한다. "Happens-before (→)" 관계라는 것을 이해하는 것이 중요하다.

A → B 는 다음과 같은 경우에 성립한다.

같은 프로세스(클라이언트)에서 A가 B보다 먼저 발생됨
A 후에 메세지를 보내고 메세지를 받은 쪽이 B를 수행함.
A → B 이고 B → C 이면 A → C 임.

Causal Consistency란 이런 원인-결과 관계가 모든 노드에서 같이 보이는 규약이다.

아래 경우는 Causal Consistency이다.

여기에서 찾을 수 있는 "Happens-before" 관계는 w(x)a → r(x)a, w(x)b → r(x)b 뿐이고 이는 모든 프로세스가 만족 한다. P3 가 b를 먼저 읽는 것은 괜찮다. 두 쓰기 연산은 독립적이므로 w(x)a → w(x)b의 원인-결과 관계는 없기 때문이다.

아래는 Causal Consistency를 성립하지 않는다. 이유는 w(x)a → w(x)c의 원인-결과 관계가 존재하는데, P3에서 c가 a보다 먼저 읽히기 때문이다.

Eventual Consistency

Eventual (결국) 일관성을 가지게 되는 규약이다. 언젠가 쓰기 연산이 없고 네트워크가 안정화 되면 결국 모든 노드는 같은 값을 보게된다. 매우 유한 규약이다.

Consistency in Replicache

Replicache는 Causal+ Consistency를 따른다고 한다. 이건 조금더 강한 Causal Consistency라는데 나중에 알아보도록하자.

Replicache

Jaeyoun Nam — Tue, 31 Dec 2024 13:25:12 +0000

Replicache

Local First Software 를 구현하는 데 도움을 주는 프레임워크이다. Git가 유사하게 push, pull 을 통해서 싱크를 맞추는 작업을 구성하는데 도움을 준다.

Replicache 는 서버 데이터의 동기화를 뒤에서 비동기적으로 수행하여 Server의 Round trip을 없애고 즉각적인 UI변경을 가능하게 한다.

Replicache Parts

Replicache는 여러개의 요소로 구성된다.

Replicache

Replicache는 내부에 git 같은 동작을 포함하는 In browser Key-Value 스토어라고 볼 수 있다. 메모리에 먼저쓰고 나중에 동기화한다.

Your application

웹 어플리케이션 같은 우리가 만든 응용프로그램이다. Replicache에 상태를 저장하는 주체이다. Mutator 와 Subscription를 구현해서 상태의 변경과 대응을 한다.

Your server

가장 신뢰할 수 있는 데이터를 저장하기 위해 존재한다. 서버에 연결된 데이터베이스에 저장된 상태는 어플리케이션에 있는 상태보다 우선시 된다.

서버는 push(upstream)와 pull(downstream)을 구현해서 클라이언트의 Replicache와 소통해야한다.

push(upstream): Replicache 는 변경 사항을 push endpoint로 보낸다. 서버에도 어플리케이션과 같이 Mutator를 구현하는데 이 push endpoint는 이 mutator를 실행시켜서 database의 상태를 변경한다.
pull(downstream): 주기적으로 혹은 명시적으로 요청할 때, Replicache는 서버에 Pull 요청을 날린다. 서버는 클라이언트가 서버의 상태와 동일해지기 위해 필요한 변경사항을 반환한다.
poke: 주기적으로 클라이언트가 pull요청을 보내긴 하지만, 좀 더 realtime으로 보여주기 위해서 서버에 변경이 있을때, 서버가 클라이언트에게 pull 요청을 하라고 힌트를 주는 신호이다. 아무런 데이터를 수반하지 않는다.

Sync

어플리케이션과 서버가 최신의 상태로 동기화 한다. 아래 그림이 이 과정을 잘 보여준다. 서버에서 주기적으로 상태변화를 pull 받아서 UI를 업데이트하는 과정, 클라이언트에서의 상태변경이 먼저 UI를 업데이트하고 서버에 push 되는 과정을 보여준다.

출처

Clients, ClientGroups, Caches

메모리에 있는 Replicache를 Client 라고 한다.

import {Replicache} from "replicache";

const rep = new Replicache({
  name: userID,
  ...
});

console.log(rep.clientID);

클라이언트는 보통 한탭 당 하나 존재한다. 클라이언트는 휘발성이고 탭의 생명주기와 함께한다. 유일한 clientID를 가진다.

Client Group은 Local data를 공유하는 클라이언트의 집합이다. 이 클라이언트 그룹 안에 있는 클라이언트들은 오프라인 상태라도 상태를 공유한다.

Client Group은 Replicache의 생성자의 name parameter에 의해 구별되는 on-disk persistent cache를 사용한다. 같은 name을 가지는 Client Group에 속하는 모든 클라이언트는 같은 캐시를 공유한다.

The Client View

Client는 ordered map of key value pair 를 persistent cache에 가지고 있는데 이를 Client View 라고 부른다. Client View는 어플리케이션의 데이터이고 서버의 데이터와 싱크된다. Client View라고 부르는 이유는 서로 다른 클라이언트 마다 서버의 데이터를 다른 Client View로 가지고 있을 수 있기 때문이다. 각 클라이언트가 서버의 상태를 보는 게 다르다는 의미다.

Client View에 접근하는 것은 매우 빠르다. Read latency 는 1ms 미만이고 대부분의 장치에서 500MB/s 의 Throughput을 가진다.

React 같은 곳에서 useState로 따로 Client View를 카피해서 메모리에 올려서 사용하지말고, Client View에서 바로 읽어서 사용하길 권장한다. Client View에 mutator가 변경을 가하면 subscription이 발동되어 UI가 업데이트 되도록 하면 된다.

Subscriptions

Subscribe 함수는 ReadTransaction 인자를 받아서 Replicache에서 읽는 것을 구현한다. Replicache의 데이터가 변해서 이 subscription이 out of date 될때마다 subscribe 함수가 다시 수행된다. 만약 이 결과가 바뀐다면, 값이 업데이트되고 UI도 업데이트 된다.

Subscription을 통해 UI를 구성하면, 항상 최신의 상태로 유지할 수 있다.

const todos = useSubscribe(rep, async tx => {
  return await tx.scan({prefix: 'todo/'}).toArray();
});
return (
  <ul>
    {todos.map(todo => (
      <li key={todo.id}>{todo.text}</li>
    ))}
  </ul>
);

Mutations

Mutation은 Replicache의 데이터를 변화시키는 작업을 뜻한다. Mutation을 받아서 실제로 데이터를 변화시키는 주체를 Mutator 라고 한다.

시작할 때 Replicache에 여러 Mutator를 등록하는데, 사실 그냥 named function이다. 아래 createTodo와 markTodoComplete는 모두 mutator로 WriteTransaction을 통해서 Replicache의 데이터를 변화시킨다.

const rep = new Replicache({
  ...
  mutators: {
    createTodo,
    markTodoComplete,
  },
});

async function createTodo(tx: WriteTransaction, todo: Todo) {
  await tx.set(`/todo/${todo.id}`, todo);
}

async function markTodoComplete(tx: WriteTransaction,
    {id, complete}: {id: string, complete: boolean}) {
  const key = `/todo/${id}`;
  const todo = await tx.get(key);
  if (!todo) {
    return;
  }
  todo.complete = complete;
  await tx.set(key, todo);
}

Mutator는 아래와 같이 작동 시킨다. Mutator가 작동하면 데이터가 변경되고, 그와 연관된 subscription들이 트리거 되면서 UI 또한 변경된다.

await rep.mutate.createTodo({id: nanoid(), text: "take out the trash"});

내부적으로 Mutator는 mutation이라는 것을 만든다. 수행 기록 같은건데, Replicache는 아래와 같은 mutation을 생성한다.

[
  {id: 1, name: "createTodo", args: {id: "t1", text: "take out the trash"}},
  {id: 2, name: "markTodoComplete", args: {id: "t1", complete: true}},
]

이 mutation들이 서버에 push 되어 완전히 sync되기 전까지 이 mutation들은 pending 상태로 표기된다.

Sync Details

이제 Replicache의 핵심이라고 할 수 있는 Sync의 자세한 내용이다. Sync는 서버에서 이루어진다.

The Replicache Sync Model

(이제부터 '상태'라는 표현은 여러 key과 value 쌍으로 된 데이터(key value space)의 상태를 뜻 한다.)

Replicache가 풀려고 하는 Sync 문제는 여러 클라이언트가 동시에 같은 상태를 변화시키는 상황이고 아래와 같은 조건을 가질 때 발생한다.

서버가 가지고 있는 상태가 source of truth이다. canonical(표준)이라고 표현한다.
클라이언트의 로컬 상태의 변화는 즉각적으로 반영된다. 이를 speculative(추측)라고 부른다.
서버는 변경사항을 정확히 한번만 적용해야하고 그 결과가 예측 가능해야한다. 서버에 적용된 변경사항이 클라이언트의 로컬 change와 합리적으로 병합될 수 있어야한다.

이 중에 마지막 항목에서 서버의 변화를 로컬 상태와 '합리적으로 병합'이라는 것은 흥미로운 주제이다. '합리적 병합'을 위해서 다음과 같은 상황들을 고려해야한다.

로컬의 변화가 아직 서버에 적용되지 않은 경우. 이 경우는 서버로부터 새로운 상태를 가져오더라도 로컬에서의 변경 사항이 앱의 UI에서 사라지지 않도록 해야 한다. 서버로부터 새로운 상태를 수신한 후, 기존의 로컬 변경 사항을 서버 상태 위에 다시 실행해야 한다.
클라이언트에서 발생한 로컬 변경 사항이 이미 서버에 전송되고, 서버 상태에 반영된 경우. 이 경우는 로컬 변경 사항을 중복 적용하는 것을 주의 해야한다. 로컬 변경 사항을 다시 적용하지 않아야한다.
동일한 상태에 대해 서버 상태를 변경한 다른 클라이언트가 존재하는 경우. 이 경우에도 첫번째 경우 처럼 서버에서 수신한 상태를 기준으로 로컬 변경 사항을 재 실행해야 한다. 하지만 같은 리소스에 대해서 충돌이 발생할 수 있기 때문에 병합 논리를 잘 짜야한다. Mutator 내부에 이 로직을 작성한다.

Mutator의 동작 과정을 따라가보자.

Local execution

로컬에서 Mutator 가 동작하고 mutator로직에 따라서 replicache의 값이 변경된다. 동시에 이 클라이언트에서 sequential하게 증가하는 mutationId를 가진 mutation을 생성한다. mutation은 pending mutation으로 queuing된다.

Push

Pending mutations은 server에 구현된 push endpoint(replicache-push)로 보내진다.

mutation은 서버에 구현된 mutator를 실행시켜서 canonical 상태를 변경한다. Mutation을 적용하면서 이 클라이언트의 last mutation id를 업데이트하고, 이 클라언트가 다음 pull을 할 때 어느 mutation부터 재 적용할지 알 수 있는 값이 된다.

로컬에 적용된 pending mutation은 speculative result를 생성하고, 서버에 적용된 mutation은 canonical result를 생성한다. 서버에 적용된 mutation은 confirmed 되어 다시 로컬에서 실행되지 않는다. 만약 같은 mutation이 다른 결과를 반환하더라도 server의 canonical result가 우선되므로 클라이언트의 결과는 바뀌게 된다.

Pull

Replicache는 최신 상태를 가져와 UI를 업데이트하기 위해서 주기적으로 pull endpoint(replicache-pull)로 요청을 보낸다.

Pull request는 cookie, clientGroupId를 담아서 요청하고, new cookie, patch, lastMutationIDChanges를 반환받는다.

cookie는 클라이언트가 가지고 있는 서버 상태를 구별하는데 사용한다. 서버와 클라이언트 상태가 얼마나 차이나는 지 추적할 수 있는 값이면 된다. 데이터베이스의 상태가 바뀔때마다 변경되는 global 'version'이라고 생각해도 된다. 아니면 좀 더 특정 범위의 데이터를 추적하는 쿠키 전략을 사용해도 된다.

lastMutationIdChanges는 각 클라이언트의 마지막으로 서버에서 적용된 mutation ID를 나타내는 값이다. 이 값보다 작은 mutationID를 가진 mutation은 모두 더 이상 pending이 아닌 confirmed로 간주해야한다.

Rebase

클라이언트가 pull을 받으면 로컬의 상태에 patch를 적용해야한다. 하지만 pending mutation이 현재 로컬 상태 영향을 줬을 것이기에 로컬 상태에 바로 patch를 적용할 수는 없다. 대신에 로컬의 pending mutation을 되돌리고 pull로 받은 patch를 먼저 적용한 후에 다시 로컬 pending mutation을 적용한다.

이런 되돌리기와 재적용을 가능하게 하기 위해서 Replicache는 깃과 유사하게 설계되었다. 서버의 상태가 main branch, 로컬에 팬딩된 mutation으로 변경된 상태를 develop 브랜치라고 생각하고 서버로 부터 main에 pull을 받고 develop을 main으로 rebase한다고 생각하면 된다.

리베이스하면서 발생할 수 있는 컨플릭트는 아래에서 따로 알아본다.

Poke

Poke는 위에서 설명했듯이 서버가 클라이언트에게 pull을 하라고 알려주는 힌트 메세지이다.

Conflict Resolution

Replicache과 같은 분산시스템에서 Merge conflict는 피할 수 없다. pull과 push과정에서 병합이 필요하다. 병합은 병합 결과가 예측 가능해야하고 앱의 목적에 맞는 방식으로 진행되어야한다.

만약 회의실 예약 앱 이라면, 컨플릭이 발생했을때 하나의 요청만 승인되어야 한다. 그렇기 때문에 먼저 예약한 클라이언트만 승인하는 병합 방법을 채택해야한다.

반대로 Todo 앱이라면, 투두리스트는 동시에 추가가 일어나더라도, 두 변경 모두 승인되는 것이 목적에 맞다.

Merge Conflict는 다음 두 상황에서 발생한다.

로컬 변경 사항이 서버에 적용될 시점. 로컬에서 적용할 때의 상태와 서버에서 적용할 때의 상태가 다를 수 있기 때문이다.
Rebase할 때. 역시 적용할때 상태가 다를 수 있기 때문이다.

Replicache는 앱의 목적에 따라 병합 방법을 다르게 구현해야 함을 인지하기 때문에, 개발자가 이를 구현할 수 있도록 한다. 개발자는 Mutator를 통해 이 로직을 구현하면 된다.

Local First Software

Jaeyoun Nam — Sat, 21 Dec 2024 07:15:07 +0000

State (상태)

웹이 점점 발전하면서 유저와 상호작용하는 요소, 보여지는 요소가 많아지게 되었다. 이런 요소 들은 유저가 보는 화면을 변화시킨다. 화면을 변화시키는 것들을 '상태'라고 정의 할 수 있다.

예를 들어서 랜딩 페이지 같은 정보성 웹페이지 인 경우에 '상태'라고 할 수 있는건 보여줄 정보 하나이다.

다음으로 깃헙 같은 경우에는 내 정보, 내 레포지토리 정보, star 수 등 다양한 정보들이 있는데 이들에 따라 유저에게 보여질 화면이 달라지기 때문에 이 들을 모두 '상태'라고 볼 수 있다.

더 복잡한 예시로, 피그마 같은 예시를 들 수 있다. 화면에 모든 점, 선, 면등 그래픽들은 모두 '상태'이다. 게다가 협업 기능은 나 이외의 다른 사람의 상태까지도 공유해야한다.

State & Data

상태는 모두 데이터다. 유저에 대한 정보, 유저 맞춤 정보 등 모두 어딘가 저장되어 있는 데이터이고 이 데이터는 곧 유저가 보는 화면의 상태가 된다. 보통 이 데이터라는 건 서버에 Single Source of Truth 로 저장된다. 어떤 웹사이트에 로그인을 했다면 그 사이트의 서버의 users 테이블에 하나의 행으로 저장될 것이다.

Data is too far

최근 웹은 복잡하다. 버튼은 셀 수 없을 만큼 많고, 한 화면에 보여주는 데이터도 많다. 시의성이 중요한 정보들도 많다. 이 상태들이 변경될 때 마다 데이터의 정합성을 위해 서버에 왔다 갔다 해야한다. 도큐먼트처럼 1분에 '다음페이지'만 받아오면 되는 경우에는 큰 문제가 되지 않는다. 하지만 노션같이 유저가 계속 데이터를 수정하는 경우라면 큰 문제가 된다. 페이지에서 특성같은 걸 설정할 때마다 로딩 해야 한다면 속이 뒤집어 질 것이다.

Optimistic Update

인스타그램 같은 소셜 미디어 사이트에서 좋아요를 누르는 것을 생각해보자. 좋아요를 누르면 서버에 가서 내가 그 포스트를 좋아요 했다는 정보를 저장하고, 그 포스트의 좋아요 숫자를 하나 올려준 다음, 현재 포스트의 좋아요를 가져와서 나한테 보여줘야 한다.

하지만 인스타그램은 0.001초 만에 애니메이션과 함께 좋아요가 눌리고 카운트가 올라간다.

이는 서버에 정보가 도달하기도 전에 클라이언트의 상태를 업데이트 하는 것으로 가능하다. 좋아요를 누른 데이터가 서버에 잘 기록되겠지 하고 클라이언트의 상태를 업데이트하는 것이다. 대부분의 경우에는 서버와의 통신이 성공할 것이기 때문에 이를 낙관적으로 성공이라고 판단하는 것이다.

물론 서버에 보낸 요청이 실패할 경우도 있기 때문에, 실패할 시 클라이언트의 상태를 롤백하는 것은 신경 써주어야한다.

Responsiveness Over Consistency

내가 좋아요를 눌렀는지 아닌지를 Optimistic 하게 보여주는 건 매우 합리적이다. 근데 내가 누를 때 다른 사람도 눌러서 좋아요가 1개 이상 늘었을 수도 있는데 이건 어떻게 처리하는 걸까.

이건 그냥 데이터 정합성을 약간 무시하는 것으로 간단하게 해결이 된다. 그 게시글이 인기 게시글이라면 내가 포스트를 보는 그 시간 동안 좋아요 개수가 안늘었을 리가 없다. 이건 그냥 그 소프트웨어의 정책이다. 빠른 응답을 위해서 약간의 데이터 정합성은 희생하는 것이다.

CAP theorem

분산 시스템 학문에는 CAP 이론이 있다. 이 이론은 분산 시스템을 구성할 때 C, A, P 중 최대 두개만 취할 수 있다는 이론이다.

C는 Consistency로 정합성이다. 어떤 노드에서 데이터를 읽던지 간에 같은 데이터를 읽어야한다.

A는 Availability로 어떤 노드가 죽더라도 모든 요청에 응답할 수 있는 가 이다.

P는 Partition-tolerance로 몇 노드의 네트워크 연결이 끊기는 경우에 동작할 수 있는가, 네트워크 연결 후에 다시 복구할 수 있는가 이다.

이 이론에 따르면 결국 CA, AP, CP 이렇게 세가지의 시스템이 가능하다.

CA

이론상은 분산 시스템이 CA를 택할 수 있지만, 네트워크 연결이 끊기면 동작하지 않는 시스템은 우리는 분산 시스템이라고 부르지 않기로 했다.

결국 분산 시스템이라면 P는 보장되어야한다.

AP

Availability Over Consistency

몇 노드의 네트워크 연결이 끊어졌을 때, 모든 노드가 그 값의 최신 상태에 대해서 동의하지 않았더라도 일단 연결되어 있는 노드의 값을 내려주는 것이다. 그렇기 때문에 끊어져있는 노드들 간에 최신 데이터는 일치하지 않을 수 있다. 하지만 사용자는 마치 최신의 데이터를 받는 것 처럼 서비스를 계속 이용할 수 있다.

대표적인 예시는 소셜미디어이다. 현실에서는 일어나지 않을 법한 일이지만, 유럽에 있는 인스타그램의 노드들과 아시아의 노들들의 네트워크 연결이 끊어져버렸다고 가정하자. 아시아에서 접근하는 유저들과 유럽에서 접근하는 유저가 보는 팔로우수, 좋아요 수 등은 이 장애기간동안 조금 달라도 괜찮다. 하지만 기능은 여전히 동작해야할 것 이다.

CP

Consistency Over Availability

네트워크 장애 상황에 최신 데이터에 대해서 확신을 할 수 없는 상황에서, 유저의 요청에 응답하지 않는 시스템이다.

예시는 보통 돈(거래)과 관련된 것들이다. 50%할인 하는 호텔 방이 하나 남은 상황에서 네트워크 단절이 일어났다고 하자. AP시스템에서는 둘 다 방이 있겠거니 하고 예약을 받아서 오버부킹을 받아버릴 가능성이 있다. CP 시스템은 이 데이터의 최신 상태에 대해서 확신할 수 없으니, 이에 대한 요청을 연기 시키거나 거부한다.

PACELC Theorem

CAP 이론은 사실 Partition에 대한 이론이다. 만약 Partition이 일어났다면 A나 C를 택해야한다는 것이다.

근데 사실 보통의 상황의 경우 Partition은 일어나지 않는다. 그런 상황에서 적용할 수 있는 이론이 PACELC 이론이다.

if (P) then (AC) else (LC)

즉, Partition인 경우에는 AC를 고려하고 아니면 LC를 고려하라는 뜻이다.

LC

Latency & Consistency

보통의 상황에서 시스템은 Latency와 Consistency를 trade off 한다. 거창하게 이론이지만 사실 이건 컴퓨터 공학 전반에 걸친 진리 같은 것이다.

Trade off를 생각한다는 것은 이 두가지 기준의 어느 정도로 타협을 본다는 것을 의미한다.

Latency는 느림에서 빠름의 정도를 직관적으로 알 수 있는데 Consistency는 어떤 정도를 가지는 지 직관적으로 알기는 힘들다.

Strong Consistency

강한 정합성은 이름만 들어도 감이 잡힌다. 그 어떤 노드로 접근하든 무조건 같은 데이터를 봐야한다. 즉 모든 노드가 같은 데이터를 가지고 있어야 가능한 Consistency이다.

은행을 생각해보면 될 것 같다.

Eventual Consistency

언젠가는 정합함 이라는 이름의 정합성이다. 어떤 변경점에 대해서 같은 시각 모든 클라이언트가 같은 값을 보지는 않지만, 동기화가 끝난 후엔 결국 같은 값을 보게 된다는 뜻이다.

따라서 소프트웨어의 특성에 따라서 Latency를 희생하면서 정합성을 챙길 것이냐 아니면 빠른 응답을 위해 정합성을 희생할 것이냐가 결정된다.

Client Server can be seen as Distributed system

일반적인 소프트웨어는 클라이언트와 서버가 있다. 이는 2개의 노드를 가진 분산 시스템으로 생각할 수도 있다. 위의 분산 시스템 이론을 적용하여 Local First Software를 생각해볼 수 있다.

Local First Software

우선 Local First Software의 특징 부터 알아보자.

Data on the local

Local First Software에서는 데이터가 멀리있어서 통신 비용이 발생하는 것을 피하기 위해서 데이터(상태)를 Local에 둔다. 유저의 인터렉션은 로컬의 데이터를 즉시 변화시키고 그에 따라서 UI도 즉시 반응한다. 심지어 로컬에 둘때 메모리에 두는것이 일반적이기 때문에 말그대로 즉시에 반응하는 소프트웨어를 만들 수 있다.

Still working on Network disconnection

Local First Software는 네트워크 단절 상태에서도 로컬에 계속 업데이트하기 때문에 여전히 작동한다. 대신 네트워크에 다시 연결 되었을 때 동기화하는 작업이 발생한다.

Apply PACELC

Network Partitioned

네트워크가 단절된 상황에서 가장 최신 정보라고 확신할 수 없어도 로컬에 있는 데이터를 활용해서 유저가 계속 사용할 수 있도록 한다. 즉 Partitioned 일때는 Availability를 택한다.

Network not Partitioned (Normal)

네트워크가 연결된 정상적인 상황에서 Local First Software는 Latency를 택한다. 로컬에 데이터를 바로 업데이트하고 서버의 데이터는 비동기적으로 처리함으로써 로컬의 데이터와 서버의 데이터가 합의(일치)가 되지 않은 상황을 용인한다. 변경내역들이 서버에 쌓이고 결국에는 모든 노드의 데이터가 동기화 된다.

Claude MCP

Jaeyoun Nam — Wed, 27 Nov 2024 08:59:48 +0000

11월 25일 엔트로픽(클로드)에서 MCP라는 것에 대한 공지가 올라왔다.

https://www.anthropic.com/news/model-context-protocol

MCP

MCP는 오픈소스 Model Context Protocol이고 이 프로토콜을 통해서 모델이 어떤 데이터(기능)에 직접 접근 할 수 있도록 해준다.

MCP 와 Tool

이전에도 Tool 이라는 개념으로 LLM이 툴 리스트를 보고 사용하면 좋을만한 툴을 골라서 사용하는 방식으로 외부 데이터에 접근할 수 있었다. 그러나 툴은 그냥 정해진 규약 없이 개발자들이 원하는 방식으로 만드는 것이라, 재사용성이 적다고 할 수 있다.

사실 MCP는 Tool의 정형화된 규약이라고 볼 수 있다. 마치 RESTFul이나 OpenTelemetry처럼 개발자들이 공통으로 사용할 규약이고 이에 맞춰서 개발하면 내껄 공유하기도 다른 사람것을 가져다 쓰기도 좋다. LLM과 데이터 소스의 통신 규약이라고 생각하면 될 것 같다.

The Protocol

MCP 프로토콜은 크게 서버, 클라이언트, 리소스로 나뉜다.

출처

클라이언트는 MCP로 서버에게 리소스를 요청하고 서버는 MCP를 보고 실제로 리소스에 접근하는 코드를 통해서 리소스를 가져온 후, 클라이언트에게 프로토콜에 명시된 형태로 리소스를 전달한다.

SQL example

도식

MySQL 데이터에 접근하는 것을 MCP 서버를 이용해서 하는 과정은 위와 같다. 누군가 MySQL과 통신해서 데이터를 가져오는 table list, query, search 등을 MCP 서버로 구현해놨다. LLM이 필요한 데이터가 MySQL에 있으면 우리는 이미 구현되어있는 MCP 서버를 내 로컬의 MySQL 데이터베이스에 붙히는것만 설정해주면 된다. 아래는 Claude Desktop App으로 LLM을 사용할 때 설정하면 되는 것이다.

{
  "mcpServers": {
    "sqlite": {
      "command": "uvx",
      "args": ["mcp-server-sqlite", "--db-path", "/Users/YOUR_USERNAME/test.db"]
    }
  }
}

이렇게 설정해주면 Claude는 MCP 서버가 어떤 일을 할 수 있는 지 알게 되고, 해당 기능이 필요할때 MCP 서버에게 요청을 보낼 수 있게 된다.

What is Good

"LLM이 데이터에 접근할 수 있게 하는 MCP"라는 멋있어보이는 타이틀이 있지만 사실 엄청 신기하고 대단한것은 아니다. 기존에 Tool로 충분히 접근하고 있던 그 방식에 권위있는 단체에서 Open된 정형화된 형식을 공표했다는데 큰 의의가 있다.

API호출로 서로 다른 서비스를 연결하는 것이 간단해졌듯이, 이런 표준화된 프로토콜은 LLM과의 통합을 더 간단하게 만들어줄 것이라고 생각한다.

Langgraph Human In The Loop with socket

Jaeyoun Nam — Wed, 13 Nov 2024 13:12:29 +0000

langgraph 의 interruption 기능을 통해서 Agent의 수행 중간에 human이 개입할 수 있다는 것을 알았다.

하지만 예시들을 보면 전부 human interaction은 한 셈치고~ 넘어간다. 실제로 User에게 확인을 받으려면 어떻게 해야할까? 크게 세가지 방법이 있을 것 같다.

Langgraph API 서버 사용

langgraph cli 로 langgraph API 서버를 docker로 실행한 후 langgraph SDK로 그래프를 실행하고, 스테이트를 변경하고, 재게하고 할 수 있다.

langgraph에서 제공하는 것들을 제공하는 방법대로 사용해야한다. 뭔간 설정이 많아지고, 내 코드랑 융합하기 까다로울 수 있어보인다.

서버에서 그래프 관리

위의 Langgraph API 서버에서 필요한 부분만 내 커스텀 서버에 구현하는 방법이다. 예를 들어 그래프 실행하면 그래프를 실행한 클라이언트와 그래프 체크포인트를 저장해야하고, 유저의 확인 후에 다시 그래프를 불러와서 유저의 응답에 맞게 상태를 변경해서 재게해야한다.

짜야할게 은근 많을 수도 있다.

소켓 연결

Agent실행 시에 소켓을 연결하고 소켓을 통해서 유저와 인터렉션 하는 것이다. 기존 예시 코드에서 소켓연결과 소켓 통신으로 유저 확인 받는 단계만 추가하면 동작한다.

대신, 글자 타이핑하듯 쳐지는 streaming을 구현하기 까다로울 수도 있다.

소켓 연결로 구현

일단 최대한 복잡성을 늘리지 않는 방향에서 구현을 해보고 싶어서 소켓연결로 구현해보았다.

서버는 NestJs를 사용하고 클라이언트는 NextJs를 사용한다.

서버

일단 Websocket 연결을 위해 Gateway를 만든다. agent/start 시에 커넥션을 만들고 바로 agent를 시행하도록 했다.

@WebSocketGateway({
  namespace: "/",
  transport: ["websocket", "polling"],
  path: "/agent/start",
  cors: {
    origin: "*",
    methods: ["GET", "POST"],
    credentials: true,
  },
})
export class AgentGateway implements OnGatewayConnection, OnGatewayDisconnect {
  @WebSocketServer()
  server: Server;
  protected readonly logger = new Logger(this.constructor.name);

  constructor(
    private readonly agentFactory: AgentFactory
  ) {}

  private pendingConfirmations = new Map<string, (response: boolean) => void>();

  // Handle new connections
  handleConnection(client: Socket) {
    console.log(`Client connected: ${client.id}`);

    // Option 1: Get actionData from query parameters
    const actionData: { agent: AgentName } = client.handshake.query.actionData
      ? JSON.parse(client.handshake.query.actionData as string)
      : null;

    if (actionData) {
      this.startAgentProcess(client, actionData);
    } else {
      // If no actionData is provided, you can wait for an event
      client.emit("error", "No action data provided");
      client.disconnect();
    }
  }

  // Handle disconnections
  handleDisconnect(client: Socket) {
    console.log(`Client disconnected: ${client.id}`);
    this.pendingConfirmations.delete(client.id);
  }

  // Send confirmation request
  async sendConfirmationRequest(clientId: string, data: any): Promise<boolean> {
    return new Promise((resolve) => {
      this.pendingConfirmations.set(clientId, resolve);
      this.server.to(clientId).emit("confirmation_request", data);

      // Optional timeout for response
      setTimeout(() => {
        if (this.pendingConfirmations.has(clientId)) {
          this.pendingConfirmations.delete(clientId);
          resolve(false); // Default to 'false' if timeout occurs
        }
      }, 3000000); // 3000 seconds timeout
    });
  }

  // Handle client's confirmation response
  @SubscribeMessage("confirmation_response")
  handleClientResponse(
    @MessageBody() data: { confirmed: boolean },
    @ConnectedSocket() client: Socket
  ) {
    const resolve = this.pendingConfirmations.get(client.id);
    if (resolve) {
      resolve(data.confirmed);
      this.pendingConfirmations.delete(client.id);
    }
  }

  // Start the agent process
  private async startAgentProcess(
    client: Socket,
    actionData: { agent: AgentName }
  ) {
    const graph = await this.agentFactory.create({
      agentName: actionData.agent,
    });

    const initialInput = { input: "hello world" };

    // Thread
    const graphStateConfig = {
      configurable: { thread_id: "1" },
      streamMode: "values" as const,
    };

    // Run the graph until the first interruption
    for await (const event of await graph.stream(
      initialInput,
      graphStateConfig
    )) {
      this.logAndEmit(client, `--- ${event.input} ---`);
    }

    // Will log when the graph is interrupted, after step 2.
    this.logAndEmit(client, "---GRAPH INTERRUPTED---");

    const userConfirmed = await this.sendConfirmationRequest(client.id, {
      message: "Do you want to proceed with this action?",
      actionData,
    });

    if (userConfirmed) {
      // If approved, continue the graph execution. We must pass `null` as
      // the input here, or the graph will
      for await (const event of await graph.stream(null, graphStateConfig)) {
        this.logAndEmit(client, `--- ${event.input} ---`);
      }
      this.logAndEmit(client, "---ACTION EXECUTED---");
    } else {
      this.logAndEmit(client, "---ACTION CANCELLED---");
    }

    // Optionally disconnect the client
    client.disconnect();
  }

  private logAndEmit(client: Socket, message: string) {
    console.log(message);
    client.emit("message", { message });
  }
}

핵심은 간단하다. Socket이 연결되면 바로 agent를 생성하여 수행하고, 수행해서 interrupt 당하면 Client에게 confirmation request message를 보내고 기다린다. confirmation이 resolve되면 이어서 graph를 진행한다.

위 코드에서 사용한 agent는 langgraph 문서에 있는 아래 스텝 1 2 3 을 순차적으로 사용하는 에이전트이다.

  const GraphState = Annotation.Root({
    input: Annotation<string>,
  });

  const step1 = (state: typeof GraphState.State) => {
    console.log("---Step 1---");
    return state;
  };

  const step2 = (state: typeof GraphState.State) => {
    console.log("---Step 2---");
    return state;
  };

  const step3 = (state: typeof GraphState.State) => {
    console.log("---Step 3---");
    return state;
  };

  const builder = new StateGraph(GraphState)
    .addNode("step1", step1)
    .addNode("step2", step2)
    .addNode("step3", step3)
    .addEdge(START, "step1")
    .addEdge("step1", "step2")
    .addEdge("step2", "step3")
    .addEdge("step3", END);

  // Set up memory
  const graphStateMemory = new MemorySaver();

  const graph = builder.compile({
    checkpointer: graphStateMemory,
    interruptBefore: ["step3"],
  });
  return graph;

클라이언트

클라이언트에서는 훅을 만들어서 agent start와 그 상태를 관리한다.

import { useRef, useState } from "react";
import io, { Socket } from "socket.io-client";

export const useAgentSocket = () => {
  const socketRef = useRef<Socket | null>(null);
  const [confirmationRequest, setConfirmationRequest] = useState<any>(null);
  const [messages, setMessages] = useState<string[]>([]);

  const connectAndRun = (actionData: any) => {
    return new Promise((resolve, reject) => {
      socketRef.current = io("http://localhost:8000", {
        path: "/agent/start",
        transports: ["websocket", "polling"],
        query: {
          actionData: JSON.stringify(actionData),
        },
      });

      socketRef.current.on("connect", () => {
        console.log("Connected:", socketRef.current?.id);
        resolve(void 0);
      });

      socketRef.current.on("connect_error", (error) => {
        console.error("Connection error:", error);
        reject(error);
      });

      // Listen for confirmation requests
      socketRef.current.on("confirmation_request", (data) => {
        setConfirmationRequest(data);
      });

      // Listen for messages
      socketRef.current.on("message", (data) => {
        console.log("Received message:", data);
        setMessages((prevMessages) => [...prevMessages, data.message]);
      });

      socketRef.current.on("disconnect", () => {
        console.log("Disconnected from server");
      });
    });
  };

  const sendConfirmationResponse = (confirmed: boolean) => {
    if (socketRef.current) {
      socketRef.current.emit("confirmation_response", { confirmed });
      setConfirmationRequest(null);
    }
  };

  const disconnectSocket = () => {
    if (socketRef.current) {
      socketRef.current.disconnect();
    }
  };

  const clearMessages = () => {
    setMessages([]);
  };

  return {
    confirmationRequest,
    sendConfirmationResponse,
    connectAndRun,
    disconnectSocket,
    messages,
    clearMessages,
  };
};

커넥션을 맺고, confirmation request가 오면 confirmationRequest 상태를 업데이트한다. UI component에서 confirmationRequest 상태를 보고 유저에게 창을 띄워주면 된다.

Langgraph - Human In the Loop

Jaeyoun Nam — Tue, 12 Nov 2024 14:33:00 +0000

Agent system 을 만들다보면 자동으로 결과까지 완벽하게 돌아가주면 좋겠지만 지금의 정확도로는 힘들다. 만약 전체 과정을 Agent가 생각하여 진행하도록 한다면, 중간에 한번의 실수만 있어도 의도와는 매우 다른 결과를 얻게 될 것이다.
그리고 Agent에게 권한을 전적으로 위임하기엔 무서운 오퍼레이션들도 있다. API콜이나, DB 수정, Bash 스크립팅등... 이 있다.

그렇기에 지금(어쩌면 꽤 오랫동안)은 Agent의 수행과정 중 인간의 간섭은 필연적이다.

Langgraph의 Human-in-the-loop

인간이 간섭하는 5가지의 상황을 지원한다.

Approval

다음 실행을 할 것인지 판단을 인간에게 맡김.

Editing

현재 상태를 유저한테 보여주고 수정할 수 있도록 함.

Input

User Input을 받는 단계를 명시하고 실제로 유저한테 값을 받는 것.

Reviewing tool calls

유저가 Tool의 결과 값을 보고 수정할 수 있도록 함.

Time travel

이전 상태(노드 수행전)로 돌아가던가, 돌아가서 다른 진행을 할 수 있도록 하는 것. Like multi-verse.

Persistant Layer

이런 인간의 간섭이 가능하게 하는 건 Langgraph의 persistant layer 덕이다. 인간의 간섭이 필요한 상태를 저장해놨다가 유저의 승인, 수정이 끝나면 다시 그 지점부터 시작이 가능하다. 게임에서 체크포인트 같은 개념이다.

Breakpoint

디버깅 툴에서 Breakpoint 설정하는 것과 같이 breakpoint는 그 지점까지만 신나게 수행하고 잠시 멈추라는 표시이다.

langgraph에서는 graph를 컴파일할때 break point를 명시할 수 있다.

graph = builder.compile(checkpointer=checkpointer, interrupt_before=["step_for_human_in_the_loop"])

Dynamic Breakpoint

컴파일 할 때 선언하는 것은 정적이다. 실행 중에 변경되는 state에 따라서 멈추는 것이 어렵다. Dynamic Breakpoint는 state에 따라서 설정할 수 있는 breakpoint이다. NodeInterrupt라는 특별한 exception이 발생하면 그래프 수행을 멈추고 유저 간섭을 기다린다.

def my_node(state: State) -> State:
    if len(state['input']) > 5:
        raise NodeInterrupt(f"Received input that is longer than 5 characters: {state['input']}")
    return state

Patterns

그림을 보면 좀 더 쉽다.

Approval

# Compile our graph with a checkpoitner and a breakpoint before the step to approve
graph = builder.compile(checkpointer=checkpoitner, interrupt_before=["node_2"])

# Run the graph up to the breakpoint
for event in graph.stream(inputs, thread, stream_mode="values"):
    print(event)

# ... Get human approval ...

# If approved, continue the graph execution from the last saved checkpoint
for event in graph.stream(None, thread, stream_mode="values"):
    print(event)

node_2 전에 그래프 수행이 끝나서 첫번째 for 구문을 탈출하게 되고, 그 후에 다시 graph.stream을 호출해서 이어서 수행한다.

Editing

# Compile our graph with a checkpoitner and a breakpoint before the step to review
graph = builder.compile(checkpointer=checkpoitner, interrupt_before=["node_2"])

# Run the graph up to the breakpoint
for event in graph.stream(inputs, thread, stream_mode="values"):
    print(event)

# Review the state, decide to edit it, and create a forked checkpoint with the new state
graph.update_state(thread, {"state": "new state"})

# Continue the graph execution from the forked checkpoint
for event in graph.stream(None, thread, stream_mode="values"):
    print(event)

역시 node_2 앞에서 멈추고, 이번에는 아무것도 안하고 다시 graph.stream을 호출하는 대신, 현재 그래프의 state를 변경한다. 그 후에 변경된 state에서 다시 그래프를 수행한다.

이외

이외의 동작들은 아래 랭그래프 공식 도큐먼트에서 확인하자

Reference

https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/

Zed에서 VSCode로 돌아옴

Jaeyoun Nam — Thu, 29 Aug 2024 09:10:58 +0000

Rust로 만들어져서 버그 프리할 것 같고, 홈페이지 들어가면 빠르다고 크게 적혀있어서 Zed를 사용해보기로함.

1차 시도

사실 1차 시도는 6달전쯤인 2024년 2월 정도에 시도해봄.

이 때 시도하다 안쓰게된 이유 중 가장 크리티컬했던건 ctrl + tab 이 이전 탭으로 이동이 아닌 것 이었다.

평생 ctrl + tab으로 이전 탭 이동하던 사람(아마 다 그럴듯)으로써 절대 적응할 수 없는 것이었기에 포기했다.

2차 시도

2024년 8월 중순 쯤 다시 깔아서 체험해봄.

위의 이슈는 ctrl + tab 이슈 다행히 해결되어 있었다.

불편한 점은 있었지만, 단축키 설정 + 손에 익히기 전략으로 약 2주간 사용했지만 다시 포기했다.

아직 불편한 점들 (2024/08/29)

심각하게 불편한 점

Find Reference 하면 reference 찾는 창이 파일 마냥 생기는데, 이 파일이 열려있는 상태에서 그 안쪽의 코드가 변경되면 따로 세이브를 해주거나 변경 안하고 저장을 해줘야함.

만약 Find Reference에 있는 코드를 삭제하고 돌아와서 이 reference를 저장해버리면, 지웠던 코드가 다시 살아남. (개불편)

Javascript 관련된 것들

import를 상대주소로 가져옴 해결법이 있는듯 하나 잘 안됨
Auto save 시 린팅이 잘 안됨

적당히 불편함 점

Vim 에서 gd 로 reference 이동이 안됨
Github extension이 없는 점

NestJS + Opentelemetry (Sampling)

Jaeyoun Nam — Sun, 18 Aug 2024 14:26:11 +0000

Grafana Cloud

이전 포스트에서 Grafana Cloud에 Opentelemetry data를 쏴서 저장하고 보는 것을 했다.

그라파나 클라우드 무료 버전을 사용하면 한달에 로그와 트레이스에 50GB정도를 준다. 유저가 얼마 없어서 Trace가 별로 안쌓이는(혹은 로그를 안찍는)서비스라면 그냥 사용해도 되겠지만, 조금 규모가 있는 상태에서 도입한다면 로그가 너무 많이 쌓여 터질까봐 두렵다.

Sampling

Sampling이란 전체에서 일부를 뽑아 쓰는 것이다. 결과적으로 저장되는 Telemetry 데이터의 수를 감소시키는 작업이다.

Why need Sampling

샘플링은 왜 필요할까?

위의 그림에서 모든 동그라미(트레이스)를 저장할 필요는 없다. 중요한 트레이스(에러, 혹은 너무 수행시간이 긴)와 전체를 대표하는 일부 표본(OK trace중 일부)만 저장하면 충분하다.

Sampling의 종류

샘플링은 크게 Head Sampling, Tail Sampling으로 나눌 수 있다.

Head Sampling

맨 앞에서 샘플링하는 것을 말한다. 대표적으로 그냥 확률적으로 샘플링 하는 것이 있다. 전체 트레이스에서 10퍼센트만 남기고 나머지는 트레이스 하지 않는 것이다.

Javascript

TraceIdRatioBasedSampler를 기본적으로 제공한다.

import { TraceIdRatioBasedSampler } from '@opentelemetry/sdk-trace-node';

const samplePercentage = 0.1;

const sdk = new NodeSDK({
  // Other SDK configuration parameters go here
  sampler: new TraceIdRatioBasedSampler(samplePercentage),
});

단점

묻고 따지지도 않고 Drop해버리는 거기 때문에 중요한 trace들이 드롭되는 경우가 있다.

Tail Sampling

뒤쪽에서 샘플링 하는 것이다. 이 때는 사용할 수 있는 정보가 많기 때문에 원하는 로직에 따라서 필터링 할 수 있다.
예를 들어, 에러 트레이스는 무조건 샘플링 하는 식이다.
보통, 콜렉터에서 일단 모든 트레이스를 받은 이후에 샘플링을 한다.

단점

구현이 어려울 수 있다. 시스템이 바뀌고 조건이 바뀌면 항상 바껴야하는 존재다.
샘플링하기 위해 Stateful인 상태를 유지하고 있어야해서 수행이 어렵다.
Tail Sampler가 vendor-specific 인 경우가 있다.

구현

Tail Sampling을 Custom Span Processor를 구현하여 구현해보자.

SamplingSpanProcessor 구현

sampling-span-processor.ts 파일 생성

import { Context } from "@opentelemetry/api";
import {
  SpanProcessor,
  ReadableSpan,
  Span,
} from "@opentelemetry/sdk-trace-node";

/**
 * Sampling span processor (including all error span and ratio of other spans)
 */
export class SamplingSpanProcessor implements SpanProcessor {
  constructor(
    private _spanProcessor: SpanProcessor,
    private _ratio: number
  ) {}

  /**
   * Forces to export all finished spans
   */
  forceFlush(): Promise<void> {
    return this._spanProcessor.forceFlush();
  }

  onStart(span: Span, parentContext: Context): void {
    this._spanProcessor.onStart(span, parentContext);
  }

  shouldSample(traceId: string): boolean {
    let accumulation = 0;
    for (let idx = 0; idx < traceId.length; idx++) {
      accumulation += traceId.charCodeAt(idx);
    }
    const cmp = (accumulation % 100) / 100;
    return cmp < this._ratio;
  }

  /**
   * Called when a {@link ReadableSpan} is ended, if the `span.isRecording()`
   * returns true.
   * @param span the Span that just ended.
   */
  onEnd(span: ReadableSpan): void {
    // Only process spans that have an error status
    if (span.status.code === 2) {
      // Status code 0 means "UNSET", 1 means "OK", and 2 means "ERROR"
      this._spanProcessor.onEnd(span);
    } else {
      if (this.shouldSample(span.spanContext().traceId)) {
        this._spanProcessor.onEnd(span);
      }
    }
  }

  /**
   * Shuts down the processor. Called when SDK is shut down. This is an
   * opportunity for processor to do any cleanup required.
   */
  async shutdown(): Promise<void> {
    return this._spanProcessor.shutdown();
  }
}

status.code가 2 (Error)거나 ratio 확률에 당첨되었을 때만 this._spanProcessor.onEnd(span); 를 호출해서 export한다.

OtelSDK 업데이트

main.ts에서 spanProcessors를 업데이트 해준다.

  spanProcessors: [
    new SamplingSpanProcessor(
      new BatchSpanProcessor(traceExporter),
      samplePercentage
    ),
  ],

NestJS + Opentelemetry (Grafana Cloud)

Jaeyoun Nam — Sun, 18 Aug 2024 12:48:54 +0000

Production환경에서 Opentelemetry 사용

Opentelemetry를 어플리케이션에 설정하고 로컬에 Otel Collector와 Loki, Tempo, Grafana를 띄워서 트레이스를 볼 수 있는 건 이 전 포스팅까지 마무리가 되었다.

이제 남은건 로컬 뿐 아니라 실제 Production 환경에서의 tracing을 보는 것이다.

그러기 위해 필요한 것은 '클라우드 상에 Log, Trace 를 저장하는 일'이다.

방법들

1. Opentelemetry Collector 를 배포하기

Opentelemetry Collector (+ Loki, Tempo 등)를 어디엔가 띄워 놓고 어플리케이션에서 쏘는 OLTP의 주소를 이 Collector로 설정하면 된다.

혹은 더 좋은 Scalability를 위해서 Load balancing을 위한 Gateway를 두고, 게이트웨이에서 OLTP를 받아서 내부의 Collectors에게 넘겨 주는 방법도 있다.

2. Grafana Alloy를 설치하여 배포하기

Grafana Alloy는 그라파나에서 제공하는 configurable한 opentelemetry collector 이다.

Docker로 배포하거나 기존에 Kubernates를 사용했다면 새로운 노드로 추가할 수 있다.

3. Collector 없이 바로 쏘기

Collector 없이 Backend(Loki, Tempo, Jaeger, 등...)로 OLTP를 바로 보내버리는 방법이다.

Backend로 Grafana Cloud의 Loki, Tempo 를 사용할 수 있어서 배포 없이 빠르게 도입 가능하다는 장점이 있다.

대신 Collector를 사용해서 얻을 수 있는 Scalability나 Processing등의 장점은 사라진다.

채택: Collector 없이 쏘기

멋지게 Collector를 배포해서 쓰고 싶으나, 기존에 Kubernetes도 쓰지 않는 환경에서 Collector따로 배포하고 설정하는데 시간이 너무 많이 소요될것 같아서 그냥 Grafana Cloud로 바로 쏘는 방법을 선택했다.

사실 실험용으로 도입해보는 거고 스타트업이어서 scalability가 크게 중요하게 작용하지 않고 (로깅이니까) 무엇보다 빠르게 해볼 수 있기 때문에, fancy하진 않지만 좋은 의사결정이다.

코드

코드 변경은 매우 심플하다. OLTP의 endpoint와 Protocol만 잘 설정해주면 된다.

Tracer



import { OTLPTraceExporter as PROTOOTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto";

const oltpTraceExporter = new PROTOOTLPTraceExporter({
  url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT + "/v1/traces",
  headers: {
    Authorization: process.env.OTEL_EXPORTER_OTLP_HEADERS_AUTHORIZATION,
  },
});

우리가 쏠 trace를 받을 endpoint(grafana cloud)는 http/protobuf 프로토콜을 받으므로 exporter-trace-otlp-proto에서 임포트해서 써야한다.

Logger



const logExporter = new OTLPLogExporter({
  url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT + "/v1/logs",
  headers: {
    Authorization: process.env.OTEL_EXPORTER_OTLP_HEADERS_AUTHORIZATION,
  },
});

Logger는 기존에도 Http protocol을 쓰고 있었기 때문에 OTLPLogExporter는 그대로 사용하면 된다.

환경변수

참고로 NestJS에서 환경 변수는 AppModule을 Initialization할 때 설정되니, AppModule 만들기 전에 완료해야하는 tracer와 logger 세팅에서 사용할 수 없다.

만약 dotenv를 사용하고 있다면 먼저 불러주어야한다.



// eslint-disable-next-line import/order

import { config } from "dotenv";

// eslint-disable-next-line import/order

import { getEnvFilePath } from "@/lib/utils/env-loader";

config(); // load env before loading tracer and logger

// eslint-disable-next-line import/order

import otelSDK from "./tracer"; // otelSDK should be imported before any other imports

// eslint-disable-next-line import/order

import createLogger from "./logger";

환경변수 값

은근 찾기 어려우니 잘 따라오자.

그라파나 접속
오른쪽 위 My Account 클릭
왼쪽 Navbar의 GRAFANA CLOUD 밑의 Stack 이름을 클릭 (없으면 new stack으로 만듬)
'Manage your stack' 화면에서 Opentelmetry 카드의 configure 누름
여기서 OTEL_EXPORTER_OTLP_ENDPOINT 를 얻을 수 있고,
아래 Password / API Token 에서 Generate Key를 눌러서 키를 만들면
Environment Variables 칸에 OTEL_EXPORTER_OTLP_HEADERS에서 Basic으로 시작하는 부분이 OTEL_EXPORTER_OTLP_HEADERS_AUTHORIZATION 변수의 값이다.

환경 변수를 등록해주고 실행한다.

그라파나 클라우드 실행

그라파나에서 Grafana Launch 를 클릭하고 Explore에서 데이터를 살펴보자