LangChain과 LangGraph로 티켓을 분류하고 답변 초안을 작성하는 지원 트리아지 에이전트
README 링크에서 발견 · 2 days ago
모델 스와핑과 휴먼 리뷰를 갖춘 상태 기반 티켓 트리아지 에이전트를 만들었습니다.
What they did
LangChain quickstart에서 시작해 init_chat_model로 채팅 모델을 표준 인터페이스 뒤에 바인딩한 뒤, 오케스트레이션을 LangGraph로 옮겨 트리아지 흐름이 단일 프롬프트가 아닌 명시적인 상태 머신이 되도록 했다. 그래프에는 인테이크 정규화, 의도 분류, 헬프센터 문서에 대한 검색, 초안 생성, 신뢰도 게이트를 위한 노드가 있었다. 조건부 엣지는 신뢰도가 낮거나 위험도가 높은 티켓(청구 분쟁, 보안 신고)을 그래프를 일시 중지하고 운영자 편집 후 재개하는 인간 검토 브랜치로 보냈다. LangChain의 모델 추상화는 교체 가능하기 때문에 노드 로직을 다시 작성하지 않고 동일한 그래프에서 두 제공자를 A/B 테스트했으며, 분류 단계에는 구조화된 출력을 사용해 다운스트림 라우팅이 산문을 파싱하는 대신 타입이 지정된 필드를 읽도록 했다. 프롬프트와 도구 정의는 버전 관리되는 파일에 두었고, 레이블이 지정된 과거 티켓으로 구성된 작은 평가 세트를 추가해 프롬프트를 변경할 때마다 실행했다. 노드별 지연 시간과 토큰 소비를 검사하기 위해 LangSmith 트레이싱을 연결했다.
How it went
그래프는 수백 건의 과거 티켓에서 엔드 투 엔드로 실행되었고, 분류 정확도는 대다수를 자동 라우팅할 수 있을 만큼 충분했으며, 리뷰 큐가 위험한 나머지를 잡아냈다. 모델 교체는 리팩터링이 아니라 설정 변경만으로 가능했고, 덕분에 eval 루프가 빨라졌다. 완성도 수준은 동작하는 내부 프로토타입이었다. 트레이스는 읽을 만했고, human-in-the-loop 일시정지/재개는 올바르게 동작했으며, 초안 품질은 에이전트 보조 응답에는 수용 가능했지만 무인 발송에는 부족했다.
Worth knowing
검색 품질은 전적으로 헬프센터 문서가 얼마나 깔끔한지에 달려 있었고, 오래된 문서는 자신감 있게 틀린 초안을 만들어냈다. 신뢰도 게이트는 보정된 확률이 아니라 수작업으로 조정한 임계값이었기 때문에 주기적으로 재조정이 필요했다. 프로덕션 배포, 인증, 속도 제한 처리는 구축되지 않았고, 평가 세트가 너무 작아서 회귀가 빠져나갈 수 있었다. 티켓당 비용과 지연 시간은 트레이스에서 대략적으로만 확인했다.
Wiring
langchain과 langgraph를 설치하세요 (uv add langchain langgraph). `init_chat_model("openai:gpt-5.5")` 또는 다른 provider 문자열로 모델을 생성하되, 교체 가능하도록 env var에서 읽으세요. 티켓 텍스트, 정규화된 필드, 분류, 검색된 문서, 초안, 신뢰도, 리뷰 상태를 담는 TypedDict 그래프 상태를 정의하세요. StateGraph로 그래프를 구성하세요: normalize, classify(타입 지정 스키마에 `with_structured_output` 사용), retrieve(docs 인덱스에 대한 retriever 또는 tool 호출), draft, gate, human_review 노드를 추가하세요. gate 이후 `add_conditional_edges`로 신뢰도와 카테고리에 따라 분기하세요. human_review interrupt가 일시정지 및 재개할 수 있도록 checkpointer와 함께 컴파일하세요. 검색과 모든 외부 조회를 LangChain tool로 감싸 모델이 호출할 수 있게 하세요. 노드별 검사를 위해 env var로 LangSmith 트레이싱을 추가하고, 레이블된 티켓을 컴파일된 그래프를 통해 재생하는 별도의 eval 스크립트를 추가하세요.Try it yourself
# Support-Ticket Triage Agent (LangChain + LangGraph)
This repo implements a triage agent with a LangGraph state machine, LangChain model access via `init_chat_model`, human-in-the-loop interrupts, versioned prompts, and a small eval harness.
## Project layout
```
triage-agent/
├── README.md
├── pyproject.toml
├── .env.example
├── src/triage_agent/
│ ├── __init__.py
│ ├── config.py
│ ├── state.py
│ ├── models.py
│ ├── prompts/
│ │ ├── __init__.py
│ │ ├── v1/
│ │ │ ├── classify.md
│ │ │ ├── draft.md
│ │ │ └── tools.yaml
│ ├── nodes/
│ │ ├── __init__.py
│ │ ├── intake.py
│ │ ├── classify.py
│ │ ├── retrieve.py
│ │ ├── draft.py
│ │ ├── gate.py
│ │ └── human_review.py
│ ├── graph.py
│ └── cli.py
├── evals/
│ ├── sample_tickets.jsonl
│ └── run_eval.py
└── tests/
└── test_graph.py
```
## `pyproject.toml`
```toml
[project]
name = "triage-agent"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
"langchain>=0.3",
"langchain-core>=0.3",
"langgraph>=0.2",
"pydantic>=2",
"python-dotenv>=1",
"langsmith>=0.1",
]
[project.optional-dependencies]
dev = ["pytest>=8"]
[project.scripts]
triage = "triage_agent.cli:main"
```
## `.env.example`
```dotenv
# Model provider (any provider supported by init_chat_model)
TRIAGE_MODEL_PROVIDER=openai
TRIAGE_MODEL_NAME=gpt-4o-mini
TRIAGE_MODEL_TEMPERATURE=0
# Optional: swap provider without code changes, e.g. anthropic / claude-3-5-sonnet-latest
# TRIAGE_MODEL_PROVIDER=anthropic
# TRIAGE_MODEL_NAME=claude-3-5-sonnet-latest
# Confidence gate
TRIAGE_CONFIDENCE_THRESHOLD=0.75
# Prompt version directory under prompts/
TRIAGE_PROMPT_VERSION=v1
# Optional LangSmith tracing
LANGCHAIN_TRACING_V2=false
LANGCHAIN_API_KEY=
LANGCHAIN_PROJECT=triage-agent
```
## `src/triage_agent/config.py`
```python
from __future__ import annotations
import os
from dataclasses import dataclass
from pathlib import Path
from dotenv import load_dotenv
load_dotenv()
PROMPTS_ROOT = Path(__file__).parent / "prompts"
@dataclass(frozen=True)
class Settings:
provider: str
model_name: str
temperature: float
confidence_threshold: float
prompt_version: str
@property
def prompt_dir(self) -> Path:
return PROMPTS_ROOT / self.prompt_version
def get_settings() -> Settings:
return Settings(
provider=os.getenv("TRIAGE_MODEL_PROVIDER", "openai"),
model_name=os.getenv("TRIAGE_MODEL_NAME", "gpt-4o-mini"),
temperature=float(os.getenv("TRIAGE_MODEL_TEMPERATURE", "0")),
confidence_threshold=float(os.getenv("TRIAGE_CONFIDENCE_THRESHOLD", "0.75")),
prompt_version=os.getenv("TRIAGE_PROMPT_VERSION", "v1"),
)
```
## `src/triage_agent/state.py`
```python
from __future__ import annotations
from typing import Literal, TypedDict
Category = Literal["billing", "security", "bug", "feature_request", "account", "other"]
class TicketState(TypedDict, total=False):
# input
raw_ticket: str
ticket_id: str
# intake
normalized_text: str
customer_id: str | None
# classification
category: Category
confidence: float
rationale: str
# retrieval
retrieved_docs: list[dict]
# draft
draft_reply: str
# gate / review
needs_human: bool
review_reason: str
operator_edit: str | None
final_reply: str
```
## `src/triage_agent/models.py`
```python
from __future__ import annotations
from langchain.chat_models import init_chat_model
from langchain_core.language_models import BaseChatModel
from .config import get_settings
def get_chat_model() -> BaseChatModel:
"""Provider is selected purely by config; no code changes needed to swap."""
s = get_settings()
return init_chat_model(
model=s.model_name,
model_provider=s.provider,
temperature=s.temperature,
)
```
## Versioned prompts — `src/triage_agent/prompts/v1/`
### `classify.md`
```markdown
You are a support-ticket classifier.
Classify the ticket into exactly one category:
- billing: invoices, charges, refunds, subscription disputes
- security: vulnerability reports, account compromise, data exposure
- bug: product malfunction, errors, crashes
- feature_request: asks for new capability
- account: login, profile, permissions
- other: anything else
Return a confidence between 0 and 1 reflecting how certain you are.
If the ticket mentions money disputes or security, still classify it, but
be conservative with confidence.
Ticket:
---
{ticket}
---
```
### `draft.md`
```markdown
You are a support agent drafting a reply.
Category: {category}
Confidence: {confidence}
Relevant documentation:
{docs}
Ticket:
---
{ticket}
---
Write a concise, empathetic reply. Do not invent policies. If information is
missing, ask a clarifying question. Do not promise refunds or security fixes.
```
### `tools.yaml`
```yaml
tools:
- name: lookup_invoice
description: Fetch invoice details by customer id.
parameters:
customer_id: {type: string, required: true}
- name: search_kb
description: Search the knowledge base for relevant articles.
parameters:
query: {type: string, required: true}
top_k: {type: integer, required: false, default: 3}
```
## `src/triage_agent/prompts/__init__.py`
```python
from __future__ import annotations
from functools import lru_cache
from pathlib import Path
import yaml
from ..config import get_settings
@lru_cache
def load_prompt(name: str) -> str:
path = get_settings().prompt_dir / f"{name}.md"
return path.read_text(encoding="utf-8")
@lru_cache
def load_tools() -> list[dict]:
path = get_settings().prompt_dir / "tools.yaml"
return yaml.safe_load(path.read_text(encoding="utf-8"))["tools"]
```
## Nodes
### `src/triage_agent/nodes/intake.py`
```python
from __future__ import annotations
import re
from ..state import TicketState
def intake_normalize(state: TicketState) -> TicketState:
text = state["raw_ticket"].strip()
text = re.sub(r"\s+", " ", text)
customer_id = None
m = re.search(r"customer[_-]?id[:=]\s*(\w+)", text, re.IGNORECASE)
if m:
customer_id = m.group(1)
return {"normalized_text": text, "customer_id": customer_id}
```
### `src/triage_agent/nodes/classify.py`
```python
from __future__ import annotations
from pydantic import BaseModel, Field
from ..models import get_chat_model
from ..prompts import load_prompt
from ..state import Category, TicketState
class Classification(BaseModel):
category: Category = Field(description="One of the allowed categories.")
confidence: float = Field(ge=0.0, le=1.0)
rationale: str = Field(description="One sentence justification.")
def classify_intent(state: TicketState) -> TicketState:
llm = get_chat_model().with_structured_output(Classification)
prompt = load_prompt("classify").format(ticket=state["normalized_text"])
result: Classification = llm.invoke(prompt)
return {
"category": result.category,
"confidence": result.confidence,
"rationale": result.rationale,
}
```
### `src/triage_agent/nodes/retrieve.py`
```python
from __future__ import annotations
from ..state import TicketState
# Placeholder KB; swap for a real retriever (vector store, MCP tool, etc.)
_KB = {
"billing": ["Refunds are processed within 5 business days after approval."],
"security": ["Report vulnerabilities to [email protected]; do not disclose publicly."],
"bug": ["Include reproduction steps and app version when reporting bugs."],
"account": ["Password resets are available from the login page."],
"feature_request": ["Feature requests are reviewed weekly by product."],
"other": ["Contact [email protected] for further help."],
}
def retrieve_docs(state: TicketState) -> TicketState:
docs = _KB.get(state["category"], [])
return {"retrieved_docs": [{"text": d} for d in docs]}
```
### `src/triage_agent/nodes/draft.py`
```python
from __future__ import annotations
from ..models import get_chat_model
from ..prompts import load_prompt
from ..state import TicketState
def draft_reply(state: TicketState) -> TicketState:
llm = get_chat_model()
docs = "\n".join(f"- {d['text']}" for d in state.get("retrieved_docs", [])) or "- (none)"
prompt = load_prompt("draft").format(
category=state["category"],
confidence=state["confidence"],
docs=docs,
ticket=state["normalized_text"],
)
msg = llm.invoke(prompt)
return {"draft_reply": msg.content if isinstance(msg.content, str) else str(msg.content)}
```
### `src/triage_agent/nodes/gate.py`
```python
from __future__ import annotations
from ..config import get_settings
from ..state import TicketState
HIGH_RISK = {"billing", "security"}
def confidence_gate(state: TicketState) -> TicketState:
s = get_settings()
reasons: list[str] = []
if state["confidence"] < s.confidence_threshold:
reasons.append(f"low confidence ({state['confidence']:.2f})")
if state["category"] in HIGH_RISK:
reasons.append(f"high-risk category ({state['category']})")
if reasons:
return {"needs_human": True, "review_reason": "; ".join(reasons)}
return {"needs_human": False, "final_reply": state["draft_reply"]}
```
### `src/triage_agent/nodes/human_review.py`
```python
from __future__ import annotations
from langgraph.types import interrupt
from ..state import TicketState
def human_review(state: TicketState) -> TicketState:
"""Interrupt the graph; resume with the operator's edited reply."""
edited = interrupt(
{
"ticket_id": state.get("ticket_id"),
"reason": state.get("review_reason"),
"draft_reply": state.get("draft_reply"),
"category": state.get("category"),
"confidence": state.get("confidence"),
}
)
# `edited` is whatever the operator supplies on resume.
return {"operator_edit": edited, "final_reply": edited}
```
## `src/triage_agent/graph.py`
```python
from __future__ import annotations
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from .nodes.classify import classify_intent
from .nodes.draft import draft_reply
from .nodes.gate import confidence_gate
from .nodes.human_review import human_review
from .nodes.intake import intake_normalize
from .nodes.retrieve import retrieve_docs
from .state import TicketState
def _route_after_gate(state: TicketState) -> str:
return "human_review" if state.get("needs_human") else END
def build_graph(checkpointer=None):
g = StateGraph(TicketState)
g.add_node("intake", intake_normalize)
g.add_node("classify", classify_intent)
g.add_node("retrieve", retrieve_docs)
g.add_node("draft", draft_reply)
g.add_node("gate", confidence_gate)
g.add_node("human_review", human_review)
g.add_edge(START, "intake")
g.add_edge("intake", "classify")
g.add_edge("classify", "retrieve")
g.add_edge("retrieve", "draft")
g.add_edge("draft", "gate")
g.add_conditional_edges("gate", _route_after_gate, {"human_review": "human_review", END: END})
g.add_edge("human_review", END)
return g.compile(checkpointer=checkpointer or MemorySaver())
```
## `src/triage_agent/cli.py`
```python
from __future__ import annotations
import argparse
import uuid
from langgraph.types import Command
from .graph import build_graph
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("ticket", help="Raw ticket text")
parser.add_argument("--ticket-id", default=None)
parser.add_argument("--edit", default=None, help="Operator edit to resume with")
args = parser.parse_args()
graph = build_graph()
thread_id = args.ticket_id or str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
result = graph.invoke(
{"raw_ticket": args.ticket, "ticket_id": thread_id},
config=config,
)
if "__interrupt__" in result:
payload = result["__interrupt__"][0].value
print("HUMAN REVIEW REQUIRED:", payload)
if args.edit is None:
print("Re-run with --edit '<reply>' to resume.")
return
result = graph.invoke(Command(resume=args.edit), config=config)
print("\nFINAL REPLY:\n", result.get("final_reply"))
if __name__ == "__main__":
main()
```
## `evals/sample_tickets.jsonl`
```jsonl
{"id":"t1","text":"I was charged twice for my subscription this month. Please refund.","expected_category":"billing"}
{"id":"t2","text":"Found an auth bypass in your API. How do I report it responsibly?","expected_category":"security"}
{"id":"t3","text":"The app crashes when I open settings on iOS 17.","expected_category":"bug"}
{"id":"t4","text":"Can you add dark mode to the dashboard?","expected_category":"feature_request"}
{"id":"t5","text":"I can't reset my password, the email never arrives.","expected_category":"account"}
{"id":"t6","text":"Just saying hi, love the product.","expected_category":"other"}
```
## `evals/run_eval.py`
```python
from __future__ import annotations
import json
import sys
from pathlib import Path
from triage_agent.graph import build_graph
DATA = Path(__file__).parent / "sample_tickets.jsonl"
def _draft_passes(draft: str) -> bool:
"""Heuristic pass check: non-empty, no placeholder tokens, reasonable length."""
if not draft or len(draft.strip()) < 20:
return False
bad = ["{", "}", "TODO", "lorem ipsum"]
return not any(b in draft for b in bad)
def main() -> int:
rows = [json.loads(line) for line in DATA.read_text().splitlines() if line.strip()]
graph = build_graph()
correct = 0
drafts_ok = 0
for row in rows:
state = graph.invoke({"raw_ticket": row["text"], "ticket_id": row["id"]})
# If interrupted, treat the draft as the candidate for pass-rate purposes.
draft = state.get("draft_reply", "")
if state.get("category") == row["expected_category"]:
correct += 1
if _draft_passes(draft):
drafts_ok += 1
n = len(rows)
print(f"Classification accuracy: {correct}/{n} = {correct / n:.2%}")
print(f"Draft pass rate: {drafts_ok}/{n} = {drafts_ok / n:.2%}")
return 0 if correct == n else 1
if __name__ == "__main__":
sys.exit(main())
```
## `tests/test_graph.py`
```python
from triage_agent.graph import build_graph
def test_graph_compiles():
assert build_graph() is not None
```
## `README.md`
```markdown
# Support-Ticket Triage Agent
LangChain for model access, LangGraph for orchestration. Classifies tickets,
retrieves docs, drafts replies, and routes low-confidence or high-risk tickets
to a human reviewer via a graph interrupt.
## Setup
```bash
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
cp .env.example .env
# edit .env with your provider credentials
```
## Environment variables
| Variable | Purpose | Default |
|---|---|---|
| `TRIAGE_MODEL_PROVIDER` | Provider passed to `init_chat_model` (e.g. `openai`, `anthropic`) | `openai` |
| `TRIAGE_MODEL_NAME` | Model name | `gpt-4o-mini` |
| `TRIAGE_MODEL_TEMPERATURE` | Sampling temperature | `0` |
| `TRIAGE_CONFIDENCE_THRESHOLD` | Below this, route to human review | `0.75` |
| `TRIAGE_PROMPT_VERSION` | Prompt directory under `prompts/` | `v1` |
| `LANGCHAIN_TRACING_V2` | Enable LangSmith tracing | `false` |
| `LANGCHAIN_API_KEY` | LangSmith API key | — |
| `LANGCHAIN_PROJECT` | LangSmith project name | `triage-agent` |
Swapping providers requires only config changes — no code edits.
## Run
```bash
triage "I was charged twice, please refund."
# If routed to review, resume with an operator edit:
triage "I was charged twice, please refund." --edit "We've issued a refund; it will appear in 5 business days."
```
## Human-in-the-loop
The `gate` node routes to `human_review` when confidence is below threshold or
the category is `billing`/`security`. `human_review` calls `interrupt(...)`,
pausing the graph. Resume with `Command(resume=<operator_reply>)`; the operator
text becomes `final_reply`.
## Prompts and tools
Prompts live in `src/triage_agent/prompts/<version>/` (`classify.md`,
`draft.md`, `tools.yaml`). Bump `TRIAGE_PROMPT_VERSION` to switch versions.
## Eval
```bash
python evals/run_eval.py
```
Reports classification accuracy and draft pass rate over `evals/sample_tickets.jsonl`.
## Tracing
Set `LANGCHAIN_TRACING_V2=true` and `LANGCHAIN_API_KEY` to send traces to
LangSmith. No code changes required.
```
## Notes on design choices
- **Structured output** uses `with_structured_output(Classification)` so the classifier returns a validated Pydantic model — no brittle JSON parsing.
- **Interrupt/resume** relies on LangGraph's `interrupt()` + `Command(resume=...)` with a `MemorySaver` checkpointer keyed by `thread_id`. Swap in a persistent checkpointer (SQLite/Postgres) for production.
- **Provider swap** is entirely driven by `init_chat_model(model=..., model_provider=...)`; the rest of the graph is provider-agnostic.
- **Eval** is intentionally small and heuristic; extend `_draft_passes` with an LLM-as-judge or rubric check when you have labeled drafts.