GLM 5.2 기반 "Claude Tag" 동등 하네스 구현 설계서
LangChain + LangGraph + Deep Agents 오픈소스 스택으로 팀 AI 구현하기
목적: Claude Tag(Anthropic, 2026-06-23)가 제공하는 팀 레벨 AI 하네스를 GLM 5.2 + 오픈소스 스택으로 자체 구축하는 방법을 설계 수준에서 상세히 기술한다.
전제: 이 문서는 아키텍처 설계와 구현 가이드이며, 실제 프로덕션 배포 전 충분한 테스트와 보안 검토가 필요하다.
작성일자: 2026-06-29
관련글
GLM 5.2는 무료인데 왜 기업들은 전환하지 못하는가?
목차
- Claude Tag 기능 분해 — 무엇을 복제해야 하는가
- 오픈소스 스택 선택 근거
- 전체 시스템 아키텍처
- 핵심 구성요소 1: GLM 5.2 LangChain 연동
- 핵심 구성요소 2: Slack Bolt 게이트웨이
- 핵심 구성요소 3: LangGraph 에이전트 그래프
- 핵심 구성요소 4: 채널별 공유 메모리
- 핵심 구성요소 5: Deep Agents 플래닝 레이어
- 핵심 구성요소 6: Ambient 모드 스케줄러
- 핵심 구성요소 7: HITL 및 거버넌스 레이어
- 도구 레지스트리 & MCP 연동
- 관찰성 & 평가 파이프라인
- 프로덕션 배포 구성
- Claude Tag 대비 한계와 트레이드오프
- 구현 로드맵 — 단계별 접근
1. Claude Tag 기능 분해
구현을 시작하기 전에 Claude Tag의 기능을 정확히 분해하는 것이 필수다. 무엇을 복제해야 하는지 모르면 아무리 좋은 도구를 써도 방향을 잃는다.
1.1 Claude Tag의 핵심 기능 목록
Claude Tag가 제공하는 기능은 다음 여섯 가지 범주로 분류된다.
① 멀티플레이어 채널 에이전트: 채널당 하나의 AI 인스턴스가 존재하며, 모든 팀원이 동일한 에이전트와 상호작용한다. 누구든 @Claude를 태그해 작업을 위임하고, 다른 팀원이 이어받을 수 있다.
② 영구적 채널 메모리: 에이전트는 채널 대화를 지속적으로 학습하여 컨텍스트를 축적한다. 매번 처음부터 설명할 필요가 없다.
③ 다단계 작업 실행: 복잡한 요청을 단계별로 분해하고, 각 단계를 순차적으로 실행하며 Slack 스레드에 진행 상황을 업데이트한다.
④ 도구 연결: 코드베이스, 데이터 소스, API, SaaS 앱 등 외부 도구에 접근하여 실제 작업을 수행한다.
⑤ Ambient(주변) 모드: 태그 없이도 채널을 모니터링하여 중요한 정보를 자발적으로 알리고, 해결되지 않은 스레드를 자동 팔로우업한다.
⑥ 장기 작업 실행: 수 시간 또는 수일에 걸친 작업을 사람의 추가 개입 없이 진행하며 중간 업데이트를 게시한다.
1.2 구현 가능성 분류
flowchart TD
subgraph POSSIBLE["오픈소스로 구현 가능"]
P1["멀티플레이어 채널 에이전트\n(LangGraph + Slack Bolt)"]
P2["채널별 공유 메모리\n(PostgreSQL + Qdrant)"]
P3["다단계 작업 분해\n(Deep Agents 플래닝)"]
P4["도구 연결\n(MCP + LangChain Tools)"]
P5["장기 작업 실행\n(LangGraph Durable Execution)"]
end
subgraph PARTIAL["부분 구현 / 추가 작업 필요"]
Q1["Ambient 모드\n(스케줄러 별도 구현)"]
Q2["자동 채널 학습\n(임베딩 파이프라인 직접 구축)"]
end
subgraph DIFFICULT["구현 어렵거나 차이 있음"]
D1["Anthropic 수준의 모델 품질\n(GLM 5.2로 대부분 커버)"]
D2["엔터프라이즈급 SSO·SCIM\n(별도 IDP 연동 필요)"]
end
2. 오픈소스 스택 선택 근거
2.1 기술 스택 전체 구성
| 역할 | 선택 기술 | 선택 이유 |
|---|---|---|
| AI 모델 | GLM 5.2 (Z.ai API / 자체 호스팅) | 분포 중심 작업에서 Claude 동등 성능, 비용 70~80% 절감 |
| LLM 통합 | LangChain ChatZhipuAI 또는 ChatOpenAI(base_url) | GLM 5.2 공식 지원, tool calling 지원 |
| 에이전트 오케스트레이션 | LangGraph v1.2+ | 상태 기계 기반, HITL, 타임트래블 디버깅, 체크포인팅 |
| 장기 작업 플래닝 | Deep Agents (LangGraph 기반) | write_todos, 서브에이전트, 파일시스템 컨텍스트 관리 |
| Slack 통합 | Slack Bolt for Python | 이벤트 기반, socket mode, 소켓 연결 내구성 |
| 단기 메모리 | LangGraph InMemorySaver → PostgresSaver | 세션 내 상태, 체크포인트 |
| 장기 채널 메모리 | Qdrant (벡터 DB) | 채널 메시지 임베딩, 유사도 검색 |
| 관계형 상태 | PostgreSQL | 채널 메타데이터, 작업 이력, 감사 로그 |
| Ambient 스케줄러 | APScheduler 또는 Temporal | 정기 채널 스캔, 미완료 작업 팔로우업 |
| 관찰성 | LangSmith 또는 Langfuse | 트레이싱, 평가, 비용 추적 |
| 컨텍스트 프로토콜 | MCP (Model Context Protocol) | 표준 도구 연결, 외부 서비스 통합 |
2.2 GLM 5.2를 선택한 구체적 이유
이 시스템에서 Claude Tag의 Claude Opus 4.8을 GLM 5.2로 교체하는 것이 합리적인 이유는 다음과 같다.
첫째, 팀 협업 채널에서 발생하는 대부분의 작업 — 태스크 요약, 초안 작성, 코드 리뷰 초안, 데이터 조회 정리 — 은 분포 중심 작업이다. GLM 5.2는 이 범주에서 Claude와 동등하거나 더 나은 성능을 보인다.
둘째, Slack 채널 에이전트 특성상 토큰 소비가 매우 높다. 채널 히스토리를 지속적으로 참조하고, Ambient 모드에서 주기적으로 채널을 스캔하기 때문이다. 이 비용을 70~80% 절감하는 것은 지속 가능성에 직결된다.
셋째, GLM 5.2는 100만 토큰 컨텍스트 윈도우를 제공하여 긴 채널 히스토리와 복잡한 코드베이스를 한 번에 처리할 수 있다.
3. 전체 시스템 아키텍처
flowchart TD
subgraph SLACK["Slack 레이어"]
USER["팀원\n(@Tag 또는 Ambient)"]
BOLT["Slack Bolt\n(이벤트 게이트웨이)"]
THREAD["Slack 스레드\n(작업 업데이트 게시)"]
end
subgraph HARNESS["하네스 레이어 (핵심)"]
DISPATCHER["이벤트 디스패처\n(채널ID → 에이전트 라우팅)"]
LANGGRAPH["LangGraph\n에이전트 그래프"]
DEEP_AGT["Deep Agents\n플래너 (write_todos)"]
ROUTER["작업 라우터\n(Center/Edge 분류)"]
end
subgraph MODEL["모델 레이어"]
GLM["GLM 5.2\n(분포 중심 작업)"]
FRONTIER["Claude Opus 4.8\n(분포 엣지 작업 폴백)"]
end
subgraph MEMORY["메모리 레이어"]
PGCHK["PostgresSaver\n(LangGraph 체크포인트)"]
QDRANT["Qdrant\n(채널 임베딩 메모리)"]
PGDB["PostgreSQL\n(작업 이력·감사 로그)"]
end
subgraph TOOLS["도구 레이어"]
MCP["MCP 서버\n(표준 프로토콜)"]
WEBTOOLS["웹 검색·코드 실행"]
APITOOLS["사내 API·ERP·BSS"]
end
subgraph OPS["운영 레이어"]
AMBIENT["Ambient 스케줄러\n(APScheduler)"]
LANGSMITH["LangSmith/Langfuse\n(관찰성)"]
HITL["HITL 게이트\n(인간 승인 포인트)"]
end
USER -->|"@tag 이벤트"| BOLT
BOLT --> DISPATCHER
DISPATCHER --> LANGGRAPH
LANGGRAPH --> DEEP_AGT
DEEP_AGT --> ROUTER
ROUTER -->|"Center 작업"| GLM
ROUTER -->|"Edge 작업"| FRONTIER
LANGGRAPH <-->|"상태 저장/복구"| PGCHK
LANGGRAPH <-->|"채널 메모리 검색"| QDRANT
LANGGRAPH <-->|"작업 기록"| PGDB
LANGGRAPH -->|"도구 실행"| MCP
MCP --> WEBTOOLS
MCP --> APITOOLS
LANGGRAPH -->|"중간 결과 게시"| THREAD
AMBIENT -->|"미완료 스레드 스캔"| LANGGRAPH
LANGGRAPH --> LANGSMITH
LANGGRAPH -->|"고위험 행동"| HITL
HITL -->|"승인 후 재개"| LANGGRAPH
3.1 데이터 흐름 요약
Slack에서 @tag 이벤트가 발생하면 Bolt가 수신하여 이벤트 디스패처로 전달한다. 디스패처는 채널 ID를 기반으로 해당 채널의 LangGraph 에이전트 인스턴스를 찾거나 신규 생성한다. LangGraph는 채널 메모리를 Qdrant에서 검색하고 체크포인트를 PostgreSQL에서 복구한 뒤 Deep Agents 플래너를 실행한다. 플래너는 복잡한 작업을 단계별로 분해하고, 각 단계는 작업 라우터를 통해 GLM 5.2 또는 Claude Opus 4.8로 전달된다. 결과는 Slack 스레드에 단계별로 게시되며, 모든 실행 내용은 LangSmith에 기록된다.
4. GLM 5.2 LangChain 연동
4.1 연동 방식 — 두 가지 옵션
GLM 5.2는 OpenAI 호환 API를 제공하므로 LangChain에서 두 가지 방식으로 연동 가능하다.
옵션 A: ChatZhipuAI (공식 통합)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 설치
# pip install langchain-community pyjwt
from langchain_community.chat_models import ChatZhipuAI
from langchain_core.messages import HumanMessage, SystemMessage
# GLM 5.2 인스턴스 생성
# 주의: ChatZhipuAI의 model 파라미터에 실제 Z.ai 플랫폼의
# GLM 5.2 모델 식별자를 사용할 것 (플랫폼 문서 확인 필요)
llm = ChatZhipuAI(
model="glm-4-5", # Z.ai 플랫폼의 실제 모델명 사용
temperature=0.1, # 재현성 높임 (팀 작업 맥락)
api_key="ZHIPUAI_API_KEY",
streaming=True,
max_tokens=4096,
)
옵션 B: ChatOpenAI with Z.ai base_url (권장 — 더 안정적)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
# 설치
# pip install langchain-openai
from langchain_openai import ChatOpenAI
# Z.ai OpenAI 호환 엔드포인트 사용
# 자체 호스팅의 경우 vLLM 엔드포인트 사용
llm_glm = ChatOpenAI(
model="glm-5.2", # 실제 모델명
base_url="https://open.bigmodel.cn/api/paas/v4/", # Z.ai API
api_key="ZHIPUAI_API_KEY",
temperature=0.1,
streaming=True,
max_tokens=4096,
)
# 자체 호스팅 시 (vLLM 기준)
llm_glm_selfhost = ChatOpenAI(
model="glm-5.2",
base_url="http://localhost:8000/v1", # vLLM 서버
api_key="not-needed", # 자체 호스팅은 불필요
temperature=0.1,
streaming=True,
)
# 엣지 작업 폴백용 Claude (선택사항)
llm_claude = ChatOpenAI(
model="claude-opus-4-8",
base_url="https://api.anthropic.com/v1",
api_key="ANTHROPIC_API_KEY",
)
중요 주의사항:
model파라미터에 사용하는 모델명은 반드시 Z.ai 플랫폼(open.bigmodel.cn) 또는 자체 호스팅 엔드포인트의 실제 모델 식별자를 확인하여 사용해야 한다. API 문서가 업데이트되면 모델명이 변경될 수 있다.
4.2 Tool Calling 검증
GLM 계열 모델의 tool calling 형식이 Claude와 다를 수 있으므로 반드시 사전 검증이 필요하다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
def search_channel_history(query: str, channel_id: str) -> str:
"""채널 히스토리에서 관련 내용을 검색한다."""
# Qdrant 검색 로직 (추후 구현)
return f"채널 {channel_id}에서 '{query}' 관련 내용 검색 결과"
def post_to_slack_thread(message: str, channel_id: str, thread_ts: str) -> str:
"""Slack 스레드에 메시지를 게시한다."""
# Slack API 호출 (추후 구현)
return "게시 완료"
# Tool binding 테스트
tools = [search_channel_history, post_to_slack_thread]
llm_with_tools = llm_glm.bind_tools(tools)
# 검증
test_response = llm_with_tools.invoke([
HumanMessage("채널 C12345에서 'API 오류' 관련 내용을 찾아줘")
])
# tool_calls 필드가 올바르게 파싱되는지 확인
assert hasattr(test_response, "tool_calls"), "Tool calling 미지원 — 모델명 또는 API 엔드포인트 확인 필요"
5. Slack Bolt 게이트웨이
5.1 이벤트 게이트웨이 구조
Slack Bolt는 Socket Mode로 동작하여 별도 퍼블릭 URL 없이 연결을 유지한다. 채널 내 @tag 이벤트와 메시지 이벤트를 모두 처리한다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
# pip install slack-bolt slack-sdk
import os
import asyncio
from slack_bolt.async_app import AsyncApp
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
from dispatcher import AgentDispatcher # 다음 섹션에서 구현
app = AsyncApp(token=os.environ["SLACK_BOT_TOKEN"])
dispatcher = AgentDispatcher()
async def handle_mention(event, say, client):
"""
채널에서 @tag 이벤트 처리.
Claude Tag의 핵심 진입점과 동일.
"""
channel_id = event["channel"]
user_id = event["user"]
text = event["text"]
thread_ts = event.get("thread_ts", event["ts"]) # 스레드 또는 새 스레드
# 즉시 수신 확인 (에이전트가 처리 시작했음을 알림)
await say(
text=f"요청을 받았습니다. 처리 중입니다... :hourglass_flowing_sand:",
thread_ts=thread_ts
)
# 에이전트 디스패처에 비동기 위임
# 채널별로 동일한 에이전트 인스턴스를 재사용 (멀티플레이어 핵심)
asyncio.create_task(
dispatcher.dispatch(
channel_id=channel_id,
user_id=user_id,
text=text,
thread_ts=thread_ts,
client=client
)
)
async def handle_message(event, client):
"""
Ambient 모드용: 채널 메시지를 메모리에 누적.
@tag 없는 일반 메시지도 채널 컨텍스트로 저장.
"""
# 봇 자신의 메시지는 무시
if event.get("bot_id"):
return
channel_id = event["channel"]
await dispatcher.index_message(
channel_id=channel_id,
message=event.get("text", ""),
user_id=event.get("user"),
ts=event["ts"]
)
async def main():
handler = AsyncSocketModeHandler(
app,
os.environ["SLACK_APP_TOKEN"]
)
await handler.start_async()
if __name__ == "__main__":
asyncio.run(main())
5.2 이벤트 디스패처 — 채널별 에이전트 라우팅
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
from typing import Dict
from langgraph_agent import ChannelAgent # 다음 섹션에서 구현
from memory import ChannelMemory # 7섹션에서 구현
class AgentDispatcher:
"""
채널 ID → 에이전트 인스턴스 매핑 관리.
Claude Tag의 '채널당 하나의 AI' 아키텍처 구현 핵심.
"""
def __init__(self):
# 채널별 에이전트 인스턴스 캐시
self._agents: Dict[str, ChannelAgent] = {}
self._memory = ChannelMemory()
def get_or_create_agent(self, channel_id: str) -> ChannelAgent:
if channel_id not in self._agents:
self._agents[channel_id] = ChannelAgent(
channel_id=channel_id,
memory=self._memory
)
return self._agents[channel_id]
async def dispatch(
self, channel_id: str, user_id: str,
text: str, thread_ts: str, client
):
agent = self.get_or_create_agent(channel_id)
# 에이전트 실행 — LangGraph 스트리밍
async for update in agent.astream(
message=text,
user_id=user_id,
thread_ts=thread_ts
):
# 각 단계 완료 시 Slack 스레드에 업데이트 게시
if update.get("type") == "step_complete":
await client.chat_postMessage(
channel=channel_id,
thread_ts=thread_ts,
text=update["content"],
unfurl_links=False
)
async def index_message(
self, channel_id: str,
message: str, user_id: str, ts: str
):
"""Ambient 모드: 채널 메시지를 벡터 DB에 인덱싱."""
await self._memory.index(
channel_id=channel_id,
content=message,
metadata={"user_id": user_id, "ts": ts}
)
6. LangGraph 에이전트 그래프
6.1 에이전트 상태 정의
LangGraph의 핵심은 명시적 상태 정의다. 채널 에이전트가 관리해야 하는 모든 상태를 타입으로 정의한다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from typing import TypedDict, Annotated, List, Optional
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
class ChannelAgentState(TypedDict):
"""
채널 에이전트의 전체 상태.
LangGraph가 체크포인트로 저장하며,
장기 실행 및 세션 간 복구에 사용된다.
"""
# 현재 대화 메시지 히스토리 (add_messages 리듀서 적용)
messages: Annotated[List[BaseMessage], add_messages]
# 채널 식별자
channel_id: str
# 요청한 사용자
user_id: str
# 현재 Slack 스레드 타임스탬프
thread_ts: str
# Deep Agents 플래너가 생성한 작업 목록
todos: List[dict]
# 현재 실행 중인 단계 인덱스
current_step: int
# 채널 메모리에서 검색된 관련 컨텍스트
retrieved_context: str
# 도구 실행 결과 누적
tool_results: List[dict]
# HITL 게이트 상태 ("pending" | "approved" | "rejected")
hitl_status: Optional[str]
# 에이전트 판단: 작업 분류 (center | edge)
task_class: Optional[str]
# 최종 응답
final_answer: Optional[str]
6.2 LangGraph 에이전트 그래프 정의
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.prebuilt import ToolNode
import psycopg
# --- 노드 함수 정의 ---
async def retrieve_context_node(state: ChannelAgentState) -> dict:
"""
Qdrant에서 채널 메모리 검색.
이전 대화와 관련된 컨텍스트를 추출한다.
"""
from memory import ChannelMemory
memory = ChannelMemory()
# 최신 사용자 메시지 추출
last_message = state["messages"][-1].content
# 벡터 유사도 검색
context = await memory.search(
channel_id=state["channel_id"],
query=last_message,
top_k=5
)
return {"retrieved_context": context}
async def classify_task_node(state: ChannelAgentState) -> dict:
"""
작업 라우터: 분포 중심 vs 엣지 분류.
분류 결과에 따라 GLM 5.2 또는 Claude Opus 선택.
"""
classifier_prompt = f"""
다음 사용자 요청이 '분포 중심(center)' 작업인지 '분포 엣지(edge)' 작업인지 판단하라.
분포 중심: 일반적인 패턴의 작업 (요약, 초안 작성, 일반 코딩, 데이터 정리 등)
분포 엣지: 전례 없는 복잡한 작업 (보안 취약점 분석, 복잡한 법률 해석, 전략 의사결정 등)
요청: {state['messages'][-1].content}
"center" 또는 "edge" 중 하나만 답하라.
"""
# GLM 5.2로 빠르게 분류 (저비용)
response = await llm_glm.ainvoke([HumanMessage(classifier_prompt)])
task_class = "center" if "center" in response.content.lower() else "edge"
return {"task_class": task_class}
async def plan_node(state: ChannelAgentState) -> dict:
"""
Deep Agents 플래너: 복잡한 작업을 단계별 TODO로 분해.
Claude Tag의 '작업을 단계로 분해' 기능 구현.
"""
from deepagents import DeepAgent # pip install deepagents
planner_message = f"""
채널 컨텍스트:
{state['retrieved_context']}
사용자 요청:
{state['messages'][-1].content}
이 요청을 실행 가능한 단계별 작업 목록으로 분해하라.
각 단계는 독립적으로 실행 가능해야 한다.
"""
# Deep Agents의 write_todos 기능 활용
# 내부적으로 LangGraph 서브그래프로 동작
llm = llm_glm if state.get("task_class") == "center" else llm_claude
todos = await deep_agent_plan(planner_message, llm)
return {
"todos": todos,
"current_step": 0
}
async def execute_step_node(state: ChannelAgentState) -> dict:
"""
현재 단계 실행 노드.
도구 호출이 필요하면 ToolNode로 분기.
"""
current_step = state["current_step"]
if current_step >= len(state["todos"]):
return {"final_answer": "모든 단계 완료"}
current_todo = state["todos"][current_step]
# 단계별 적합 모델 선택
llm = llm_glm if state.get("task_class") == "center" else llm_claude
llm_with_tools = llm.bind_tools(all_tools)
step_prompt = f"""
채널 컨텍스트: {state['retrieved_context']}
이전 단계 결과: {state.get('tool_results', [])}
현재 단계: {current_todo['description']}
이 단계를 실행하라. 필요한 도구를 사용하되, 결과를 명확히 보고하라.
"""
response = await llm_with_tools.ainvoke(
state["messages"] + [HumanMessage(step_prompt)]
)
return {
"messages": [response],
"current_step": current_step + 1
}
async def synthesize_node(state: ChannelAgentState) -> dict:
"""최종 결과 통합 및 Slack 게시 형식으로 포맷."""
synthesis_prompt = f"""
완료된 작업들의 결과를 통합하여 Slack에 게시할 명확한 응답을 작성하라.
원래 요청: {state['messages'][0].content}
실행 결과: {state.get('tool_results', [])}
응답은 간결하고 실행 가능한 형태로 작성하라.
"""
llm = llm_glm if state.get("task_class") == "center" else llm_claude
response = await llm.ainvoke([HumanMessage(synthesis_prompt)])
return {"final_answer": response.content}
# --- 조건부 엣지 함수 ---
def should_use_hitl(state: ChannelAgentState) -> str:
"""
HITL 게이트 판단: 고위험 작업인지 확인.
파일 삭제, 외부 API 쓰기 등은 인간 승인 필요.
"""
if state.get("hitl_status") == "pending":
return "wait_for_approval"
# 위험한 도구 호출이 포함된 경우
high_risk_tools = {"delete_file", "send_email", "deploy_code"}
planned_tools = {
t for step in state.get("todos", [])
for t in step.get("required_tools", [])
}
if planned_tools & high_risk_tools:
return "request_approval"
return "execute"
def step_complete_or_done(state: ChannelAgentState) -> str:
"""단계가 남아있는지 확인."""
if state["current_step"] >= len(state.get("todos", [])):
return "synthesize"
return "execute_step"
# --- 그래프 조립 ---
def build_channel_agent_graph():
graph = StateGraph(ChannelAgentState)
# 노드 등록
graph.add_node("retrieve_context", retrieve_context_node)
graph.add_node("classify_task", classify_task_node)
graph.add_node("plan", plan_node)
graph.add_node("hitl_gate", hitl_gate_node) # 10섹션에서 구현
graph.add_node("execute_step", execute_step_node)
graph.add_node("tools", ToolNode(all_tools)) # 도구 실행
graph.add_node("synthesize", synthesize_node)
# 엣지 정의
graph.set_entry_point("retrieve_context")
graph.add_edge("retrieve_context", "classify_task")
graph.add_edge("classify_task", "plan")
graph.add_conditional_edges(
"plan",
should_use_hitl,
{
"execute": "execute_step",
"request_approval": "hitl_gate",
"wait_for_approval": "hitl_gate"
}
)
graph.add_conditional_edges(
"execute_step",
step_complete_or_done,
{
"execute_step": "execute_step", # 루프
"synthesize": "synthesize"
}
)
graph.add_edge("tools", "execute_step")
graph.add_edge("synthesize", END)
return graph
6.3 PostgreSQL 체크포인터 연결
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import psycopg
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
async def create_agent_with_checkpointing():
"""
PostgreSQL 기반 체크포인터로 장기 실행 에이전트 구성.
세션이 종료되어도 상태가 보존되며 재개 가능.
"""
conn = await psycopg.AsyncConnection.connect(
os.environ["POSTGRES_URI"],
autocommit=True
)
checkpointer = AsyncPostgresSaver(conn)
# 체크포인트 테이블 초기화 (최초 1회)
await checkpointer.setup()
graph = build_channel_agent_graph()
compiled = graph.compile(checkpointer=checkpointer)
return compiled
7. 채널별 공유 메모리
Claude Tag가 “채널을 팔로우하며 컨텍스트를 축적”하는 기능의 핵심이다.
7.1 메모리 아키텍처 — 3계층
flowchart LR
subgraph SHORT["단기 메모리 (In-Session)"]
MSG["현재 대화 메시지\n(LangGraph State)"]
CTX["검색된 컨텍스트\n(retrieved_context)"]
end
subgraph MID["중기 메모리 (Cross-Session)"]
PGCHECK["LangGraph 체크포인트\n(PostgreSQL)"]
end
subgraph LONG["장기 메모리 (Channel Memory)"]
EMBED["채널 메시지 임베딩\n(Qdrant)"]
META["메타데이터 인덱스\n(PostgreSQL)"]
end
MSG --> PGCHECK
PGCHECK -->|"세션 재개 시"| MSG
EMBED -->|"유사도 검색"| CTX
META -->|"시간·사용자 필터"| EMBED
7.2 채널 메모리 구현
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
# pip install qdrant-client langchain-qdrant sentence-transformers
from qdrant_client import AsyncQdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from langchain_openai import OpenAIEmbeddings
from langchain_qdrant import QdrantVectorStore
import uuid
class ChannelMemory:
"""
채널별 벡터 메모리.
Slack 메시지를 임베딩하여 에이전트가 관련 히스토리를 검색할 수 있게 한다.
Claude Tag의 '채널을 팔로우하며 학습' 기능 구현.
"""
def __init__(self):
self.client = AsyncQdrantClient(
url=os.environ.get("QDRANT_URL", "http://localhost:6333")
)
# 임베딩 모델: GLM 5.2 API 임베딩 또는 오픈소스 모델 사용
# 비용 절감을 위해 sentence-transformers 로컬 모델 권장
self.embeddings = OpenAIEmbeddings(
model="text-embedding-3-small",
# 또는 자체 임베딩 서버 사용:
# base_url="http://localhost:8001/v1"
)
self._collections_initialized = set()
async def _ensure_collection(self, channel_id: str):
"""채널별 컬렉션 초기화 (최초 1회)."""
collection_name = f"channel_{channel_id}"
if collection_name not in self._collections_initialized:
collections = await self.client.get_collections()
existing = [c.name for c in collections.collections]
if collection_name not in existing:
await self.client.create_collection(
collection_name=collection_name,
vectors_config=VectorParams(
size=1536, # text-embedding-3-small 차원
distance=Distance.COSINE
)
)
self._collections_initialized.add(collection_name)
return collection_name
async def index(
self, channel_id: str, content: str,
metadata: dict = None
):
"""채널 메시지를 벡터 DB에 인덱싱."""
if not content or len(content.strip()) < 5:
return # 너무 짧은 메시지는 건너뜀
collection_name = await self._ensure_collection(channel_id)
# 임베딩 생성
vector = await self.embeddings.aembed_query(content)
await self.client.upsert(
collection_name=collection_name,
points=[PointStruct(
id=str(uuid.uuid4()),
vector=vector,
payload={
"content": content,
"channel_id": channel_id,
"timestamp": metadata.get("ts") if metadata else None,
"user_id": metadata.get("user_id") if metadata else None,
}
)]
)
async def search(
self, channel_id: str, query: str, top_k: int = 5
) -> str:
"""채널 히스토리에서 관련 컨텍스트 검색."""
collection_name = await self._ensure_collection(channel_id)
query_vector = await self.embeddings.aembed_query(query)
results = await self.client.query_points(
collection_name=collection_name,
query=query_vector,
limit=top_k,
score_threshold=0.7 # 관련성 임계치
)
if not results.points:
return "관련 채널 히스토리 없음"
context_parts = []
for point in results.points:
payload = point.payload
context_parts.append(
f"[{payload.get('timestamp', '')}] "
f"{payload.get('user_id', '알 수 없음')}: "
f"{payload.get('content', '')}"
)
return "\n".join(context_parts)
8. Deep Agents 플래닝 레이어
Deep Agents는 LangGraph 위에 구축된 고수준 패키지로, Claude Code/Manus와 동일한 아키텍처를 제공한다. write_todos 도구를 통해 복잡한 작업을 단계별로 분해하고, 파일시스템 기반 컨텍스트 관리로 긴 작업을 수행한다.
8.1 Deep Agents 설치 및 설정
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# pip install deepagents
from deepagents import DeepAgent, DeepAgentConfig
async def deep_agent_plan(task_description: str, llm) -> list:
"""
Deep Agents의 플래닝 기능을 사용하여
복잡한 작업을 실행 가능한 단계로 분해.
내부적으로 write_todos 도구를 사용하며,
LangGraph 서브그래프로 동작한다.
"""
config = DeepAgentConfig(
model=llm,
max_steps=20,
enable_subagents=True, # 병렬 서브에이전트 허용
filesystem_context=True, # 파일시스템 컨텍스트 관리
planning_mode="explicit", # 명시적 플래닝 단계 실행
)
agent = DeepAgent(config=config)
# 플래닝만 실행 (실제 실행은 LangGraph 메인 그래프에서)
plan = await agent.plan(task_description)
# 반환 형식: [{"step": 1, "description": "...", "required_tools": [...]}]
return plan.todos
8.2 SKILL.md 패턴 적용
Deep Agents는 SKILL.md 파일을 통해 특화된 도메인 지식을 동적으로 로드한다. 회사 고유의 업무 프로세스를 SKILL.md로 정의하면 에이전트가 자동으로 적용한다.
skills/slack-channel-agent/SKILL.md
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
---
name: slack-channel-agent
description: >
Slack 채널에서 팀 협업 작업을 처리한다. 요약, 작업 추적,
미팅 노트 작성, 팀 알림에 사용. 코드 작성이나 시스템 변경에는 사용하지 말 것.
license: MIT
allowed-tools:
- search_channel_history
- post_to_slack_thread
- read_confluence_page
- search_jira_issues
---
# Slack Channel Agent Skill
## 핵심 원칙
1. 작업을 위임받으면 먼저 채널 히스토리를 검색하여 관련 맥락을 파악한다
2. 완료 가능한 작업만 수행하고, 불확실한 경우 팀에 명확화를 요청한다
3. 각 단계 완료 시 스레드에 진행 상황을 업데이트한다
4. 민감한 정보(인사, 재무)는 공개 채널에 게시하지 않는다
## 작업 처리 순서
1. `search_channel_history` 로 맥락 파악
2. 작업 복잡도 판단 → 단순하면 즉시 처리, 복잡하면 계획 수립
3. 단계별 실행 및 스레드 업데이트
4. 최종 결과 요약 게시
9. Ambient 모드 스케줄러
Claude Tag의 Ambient 모드는 태그 없이도 에이전트가 자발적으로 채널을 모니터링하고 업데이트하는 기능이다. APScheduler로 구현한다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
# pip install apscheduler
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger
import datetime
class AmbientScheduler:
"""
Claude Tag의 Ambient 모드 구현.
정기적으로 채널을 스캔하여 자발적 업데이트 제공.
"""
def __init__(self, dispatcher: AgentDispatcher, slack_client):
self.dispatcher = dispatcher
self.slack_client = slack_client
self.scheduler = AsyncIOScheduler()
self._monitored_channels: dict = {} # {channel_id: config}
def enable_for_channel(self, channel_id: str, config: dict):
"""특정 채널에 Ambient 모드 활성화."""
self._monitored_channels[channel_id] = config
def start(self):
# 미완료 스레드 팔로우업: 4시간마다
self.scheduler.add_job(
self._followup_unresolved_threads,
IntervalTrigger(hours=4),
id="followup_threads"
)
# 일일 채널 요약: 매일 오전 9시
self.scheduler.add_job(
self._daily_channel_summary,
"cron", hour=9, minute=0,
id="daily_summary"
)
self.scheduler.start()
async def _followup_unresolved_threads(self):
"""
해결되지 않은 스레드를 찾아 자동 팔로우업.
Claude Tag의 '조용히 사라진 스레드 팔로우업' 기능.
"""
for channel_id, config in self._monitored_channels.items():
if not config.get("ambient_enabled", False):
continue
# 48시간 이상 응답 없는 스레드 검색
cutoff = datetime.datetime.now() - datetime.timedelta(hours=48)
unresolved = await self._find_unresolved_threads(
channel_id, since=cutoff
)
for thread in unresolved[:3]: # 한 번에 최대 3개
await self.dispatcher.dispatch(
channel_id=channel_id,
user_id="ambient_bot",
text=f"이 스레드가 {thread['age_hours']}시간 동안 미해결 상태입니다. "
f"현재 상황을 확인하고 팔로우업이 필요한지 판단하세요.",
thread_ts=thread["ts"],
client=self.slack_client
)
async def _daily_channel_summary(self):
"""매일 아침 채널 주요 내용 요약 게시."""
for channel_id, config in self._monitored_channels.items():
if not config.get("daily_summary", False):
continue
yesterday = datetime.datetime.now() - datetime.timedelta(days=1)
# 어제 채널 메시지 검색 및 요약 생성
await self.dispatcher.dispatch(
channel_id=channel_id,
user_id="ambient_bot",
text="어제 이 채널에서 논의된 주요 내용을 3-5개 항목으로 요약하고, "
"오늘 팀이 주목해야 할 사항을 추가하세요.",
thread_ts=str(yesterday.timestamp()),
client=self.slack_client
)
10. HITL 및 거버넌스 레이어
고위험 작업에 대한 인간 승인 게이트는 기업 환경에서 필수적이다. LangGraph의 interrupt 메커니즘을 활용한다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
from langgraph.types import interrupt, Command
async def hitl_gate_node(state: ChannelAgentState) -> Command:
"""
HITL(Human-in-the-Loop) 게이트.
고위험 작업 실행 전 팀 승인 요청.
LangGraph interrupt로 실행을 일시 정지하고
승인 후 자동 재개.
"""
# 계획된 고위험 작업 목록
high_risk_actions = [
step for step in state.get("todos", [])
if any(
t in step.get("required_tools", [])
for t in {"delete_file", "send_email", "deploy_code", "modify_database"}
)
]
if not high_risk_actions:
return Command(goto="execute_step")
# Slack에 승인 요청 메시지 게시
approval_message = (
f"⚠️ *승인 필요*\n\n"
f"다음 고위험 작업을 실행하려 합니다:\n"
+ "\n".join([f"• {a['description']}" for a in high_risk_actions])
+ "\n\n✅ 승인: `@tag approve` / ❌ 거부: `@tag reject`"
)
# interrupt()로 실행 일시 정지 — LangGraph가 상태를 보존
# 사용자 응답이 올 때까지 대기
approval_result = interrupt({"message": approval_message})
if approval_result.get("decision") == "approved":
return Command(
goto="execute_step",
update={"hitl_status": "approved"}
)
else:
return Command(
goto=END,
update={
"hitl_status": "rejected",
"final_answer": "작업이 팀에 의해 거부되었습니다."
}
)
# Slack에서 승인/거부 이벤트 처리
async def handle_approval(message, say, client):
"""팀원의 승인/거부 응답 처리."""
channel_id = message["channel"]
thread_ts = message.get("thread_ts")
text = message["text"].lower()
if not thread_ts:
return
decision = "approved" if "approve" in text else "rejected"
# 중단된 LangGraph 실행 재개
agent = dispatcher.get_or_create_agent(channel_id)
await agent.resume(
thread_ts=thread_ts,
input={"decision": decision}
)
11. 도구 레지스트리 & MCP 연동
11.1 기본 도구 세트
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
from langchain_core.tools import tool
from langchain_community.tools import TavilySearchResults
# 웹 검색
web_search = TavilySearchResults(max_results=3)
async def search_channel_history(query: str, channel_id: str) -> str:
"""채널 히스토리에서 관련 내용 검색."""
memory = ChannelMemory()
return await memory.search(channel_id=channel_id, query=query)
async def read_confluence_page(page_id: str) -> str:
"""Confluence 페이지 내용 읽기."""
# Confluence REST API 호출
import httpx
async with httpx.AsyncClient() as client:
response = await client.get(
f"{os.environ['CONFLUENCE_URL']}/rest/api/content/{page_id}",
auth=(os.environ['CONFLUENCE_USER'], os.environ['CONFLUENCE_TOKEN'])
)
return response.json().get("body", {}).get("storage", {}).get("value", "")
async def create_jira_ticket(
summary: str, description: str, project_key: str
) -> str:
"""Jira 티켓 생성."""
# Jira REST API 호출
import httpx
async with httpx.AsyncClient() as client:
response = await client.post(
f"{os.environ['JIRA_URL']}/rest/api/2/issue",
json={
"fields": {
"project": {"key": project_key},
"summary": summary,
"description": description,
"issuetype": {"name": "Task"}
}
},
auth=(os.environ['JIRA_USER'], os.environ['JIRA_TOKEN'])
)
return f"Jira 티켓 생성 완료: {response.json().get('key')}"
# 전체 도구 목록
all_tools = [
web_search,
search_channel_history,
read_confluence_page,
create_jira_ticket,
# 회사 고유 도구 추가
]
11.2 MCP 서버 연동
표준 MCP 프로토콜을 사용하면 도구를 중앙에서 관리하고 여러 에이전트가 공유할 수 있다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# pip install mcp langchain-mcp-adapters
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
async def load_mcp_tool_set(server_config: dict) -> list:
"""
MCP 서버에서 도구를 동적으로 로드.
새 도구 추가 시 코드 변경 없이 MCP 서버에만 추가.
"""
server_params = StdioServerParameters(
command=server_config["command"],
args=server_config.get("args", []),
env=server_config.get("env")
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await load_mcp_tools(session)
return tools
# MCP 서버 목록 (환경별 설정)
MCP_SERVERS = {
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"env": {"GITHUB_PERSONAL_ACCESS_TOKEN": os.environ.get("GITHUB_TOKEN")}
},
"slack": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-slack"],
"env": {"SLACK_BOT_TOKEN": os.environ.get("SLACK_BOT_TOKEN")}
},
# 사내 API MCP 서버 추가
}
12. 관찰성 & 평가 파이프라인
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
# pip install langsmith
import langsmith
from langsmith import traceable
# LangSmith 설정
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "LANGSMITH_API_KEY"
os.environ["LANGCHAIN_PROJECT"] = "channel-agent-glm52"
# 비용 추적 커스텀 콜백
from langchain_core.callbacks import BaseCallbackHandler
class CostTrackingCallback(BaseCallbackHandler):
"""
GLM 5.2 vs Claude 토큰 비용 추적.
모델 전환 효과를 측정하는 핵심 지표.
"""
# GLM 5.2 가격 (추정치 — 실제 Z.ai 공식 가격 확인 필요)
PRICE_PER_1K = {
"glm-5.2": {"input": 0.0003, "output": 0.0006},
"claude-opus-4-8": {"input": 0.015, "output": 0.075},
}
def __init__(self):
self.session_costs = {}
def on_llm_end(self, response, **kwargs):
model = kwargs.get("invocation_params", {}).get("model", "unknown")
usage = response.llm_output.get("token_usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
prices = self.PRICE_PER_1K.get(model, {"input": 0, "output": 0})
cost = (
input_tokens / 1000 * prices["input"]
+ output_tokens / 1000 * prices["output"]
)
if model not in self.session_costs:
self.session_costs[model] = 0
self.session_costs[model] += cost
def get_total_savings(self) -> dict:
"""GLM 5.2 사용으로 절감된 비용 계산."""
glm_cost = self.session_costs.get("glm-5.2", 0)
claude_equivalent = glm_cost / 0.02 * 0.075 # 대략적 비교
return {
"actual_cost": glm_cost,
"claude_equivalent": claude_equivalent,
"savings": claude_equivalent - glm_cost,
"savings_pct": (1 - glm_cost / max(claude_equivalent, 0.0001)) * 100
}
13. 프로덕션 배포 구성
13.1 Docker Compose 구성
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
# docker-compose.yml
version: '3.9'
services:
# 채널 에이전트 메인 서비스
channel-agent:
build: .
environment:
- SLACK_BOT_TOKEN=${SLACK_BOT_TOKEN}
- SLACK_APP_TOKEN=${SLACK_APP_TOKEN}
- ZHIPUAI_API_KEY=${ZHIPUAI_API_KEY}
- POSTGRES_URI=postgresql://agent:secret@postgres:5432/agentdb
- QDRANT_URL=http://qdrant:6333
- LANGCHAIN_API_KEY=${LANGSMITH_API_KEY}
depends_on:
- postgres
- qdrant
restart: unless-stopped
deploy:
replicas: 2 # 고가용성
# 상태 저장소
postgres:
image: postgres:16
environment:
POSTGRES_DB: agentdb
POSTGRES_USER: agent
POSTGRES_PASSWORD: secret
volumes:
- postgres_data:/var/lib/postgresql/data
# 벡터 메모리
qdrant:
image: qdrant/qdrant:latest
volumes:
- qdrant_data:/qdrant/storage
ports:
- "6333:6333"
volumes:
postgres_data:
qdrant_data:
13.2 요구사항 파일
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# requirements.txt — 주요 패키지 (버전 고정 필수)
langchain>=0.3.0
langchain-community>=0.3.0
langchain-openai>=0.2.0
langgraph>=1.2.0
deepagents>=0.1.0
slack-bolt>=1.19.0
qdrant-client>=1.9.0
langchain-qdrant>=0.2.0
psycopg[binary]>=3.1.0
apscheduler>=3.10.0
langsmith>=0.1.0
mcp>=1.0.0
langchain-mcp-adapters>=0.1.0
httpx>=0.27.0
pyjwt>=2.8.0
14. Claude Tag 대비 한계와 트레이드오프
이 시스템을 구현하기 전에 Claude Tag와의 차이를 정직하게 이해해야 한다.
flowchart LR
subgraph ADVANTAGE["자체 구축 우위"]
A1["비용 70~80% 절감\n(GLM 5.2 활용)"]
A2["컨텍스트 소유권\n(데이터 자사 보관)"]
A3["커스터마이징 자유\n(Skills·도구 완전 제어)"]
A4["벤더 잠금 없음\n(모델 교체 자유)"]
end
subgraph DISADVANTAGE["Claude Tag 우위"]
D1["구현 복잡도\n(상당한 엔지니어링 필요)"]
D2["유지보수 부담\n(내부 운영 필요)"]
D3["모델 품질 차이\n(분포 엣지 작업)"]
D4["즉시 사용 불가\n(설정·테스트 기간 필요)"]
end
14.1 반드시 알아야 할 구현 어려움
GLM 5.2 tool calling 형식 차이: GLM 계열 모델의 tool calling 응답 형식이 Claude와 미묘하게 다를 수 있다. bind_tools() 호출 후 반드시 실제 응답 파싱을 검증해야 하며, 파싱 오류 처리 로직이 필요하다.
Slack Socket Mode의 타임아웃 제약: Slack은 app_mention 이벤트에 3초 내 응답을 요구한다. 장기 실행 에이전트 작업은 반드시 비동기로 처리하고 즉시 수신 확인(ACK)을 보내야 한다. 위 코드에서는 asyncio.create_task()로 처리했다.
채널 메모리 인덱싱 비용: 모든 Slack 메시지를 임베딩하는 것은 상당한 API 비용이 발생한다. 관련성이 낮은 짧은 메시지(이모지 반응 등)는 필터링하거나, 자체 임베딩 모델(sentence-transformers)을 사용해 비용을 절감해야 한다.
LangGraph 버전 변경 속도: LangGraph의 API가 빠르게 변경된다. 2026년 1월~5월 사이에만 세 번의 마이너 버전이 출시되었다. requirements.txt에 버전을 반드시 고정하고, 버전 업그레이드 전 충분한 테스트가 필요하다.
15. 구현 로드맵
전체 시스템을 한 번에 구축하려 하면 실패한다. 단계별로 접근하는 것이 현실적이다.
flowchart LR
subgraph PHASE1["1단계 (2~3주)\n기본 채널 에이전트"]
P1A["Slack Bolt 연동\n@tag 이벤트 수신"]
P1B["GLM 5.2 LangChain 연동\n기본 응답 확인"]
P1C["단순 도구 2~3개\n(웹 검색, 채널 검색)"]
end
subgraph PHASE2["2단계 (3~4주)\n상태 관리 & 메모리"]
P2A["PostgreSQL 체크포인팅\n세션 간 상태 유지"]
P2B["Qdrant 채널 메모리\n메시지 인덱싱"]
P2C["LangGraph 다단계 실행\n스레드 업데이트"]
end
subgraph PHASE3["3단계 (4~6주)\n고급 기능"]
P3A["Deep Agents 플래닝\n복잡 작업 분해"]
P3B["HITL 게이트\n고위험 행동 승인"]
P3C["작업 라우터\nCenter/Edge 분류"]
end
subgraph PHASE4["4단계 (2~3주)\n프로덕션화"]
P4A["Ambient 스케줄러\n자발적 업데이트"]
P4B["관찰성 파이프라인\n비용·품질 추적"]
P4C["거버넌스·감사 로그\n컴플라이언스"]
end
PHASE1 --> PHASE2
PHASE2 --> PHASE3
PHASE3 --> PHASE4
단계별 성공 기준
1단계 완료 기준: @tag로 팀원이 질문하면 GLM 5.2가 채널 컨텍스트를 이해하고 스레드에 답변한다.
2단계 완료 기준: 서버 재시작 후에도 이전 대화를 기억하고 이어서 응답한다. 채널 히스토리를 검색해 관련 내용을 찾는다.
3단계 완료 기준: “이번 분기 KPI를 분석하고 Jira 티켓을 생성해줘”처럼 다단계가 필요한 요청을 단계별로 분해하여 실행한다.
4단계 완료 기준: 태그 없이도 미완료 스레드를 팔로우업하고, 일일 채널 요약을 자동 게시하며, 모든 에이전트 행동이 감사 로그에 기록된다.
참고 자료
- LangGraph 공식 문서: https://langchain-ai.github.io/langgraph/
- Deep Agents 공식 문서: https://langchain.com/resources/ai-agent-frameworks (Deep Agents 섹션)
- Slack Bolt for Python: https://slack.dev/bolt-python/
- Z.ai API 공식 문서 (GLM 5.2): https://docs.z.ai/
- ChatZhipuAI LangChain 통합: https://python.langchain.com/docs/integrations/chat/zhipuai
- Qdrant 공식 문서: https://qdrant.tech/documentation/
- LangGraph PostgresSaver: https://langchain-ai.github.io/langgraph/reference/checkpoints/
- MCP LangChain 어댑터: https://github.com/langchain-ai/langchain-mcp-adapters
- Hermes Agent (참고용 오픈소스 하네스): https://hermes-agent.nousresearch.com/
작성일자: 2026-06-29