Initial implementation of Telegram AI customer support bot
Cloudflare Workers + Hono + D1 + KV + R2 stack with 4 specialized AI agents (onboarding, troubleshoot, asset, billing), OpenAI function calling with 7 tool definitions, human escalation, pending action approval workflow, feedback collection, audit logging, i18n (ko/en), and Workers AI fallback. 43 source files, 45 tests passing. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
kappa
371
src/routes/handlers/message-handler.ts
Normal file
371
src/routes/handlers/message-handler.ts
Normal file
@@ -0,0 +1,371 @@
|
||||
/**
|
||||
* Message Handler - 텔레그램 메시지 처리의 핵심 모듈
|
||||
*
|
||||
* 메시지 수신 -> 사용자 등록/확인 -> 에이전트 라우팅 -> AI 응답 -> 피드백 수집
|
||||
*/
|
||||
|
||||
import { sendMessage, sendMessageWithKeyboard, sendChatAction } from '../../telegram';
|
||||
import { checkRateLimit } from '../../security';
|
||||
import { registerAgent, routeToActiveAgent } from '../../agents/agent-registry';
|
||||
import { OnboardingAgent } from '../../agents/onboarding-agent';
|
||||
import { TroubleshootAgent } from '../../agents/troubleshoot-agent';
|
||||
import { AssetAgent } from '../../agents/asset-agent';
|
||||
import { BillingAgent } from '../../agents/billing-agent';
|
||||
import { getMessage } from '../../i18n';
|
||||
import {
|
||||
ONBOARDING_PATTERNS,
|
||||
TROUBLESHOOT_PATTERNS,
|
||||
ASSET_PATTERNS,
|
||||
BILLING_PATTERNS,
|
||||
} from '../../utils/patterns';
|
||||
import { selectToolsForMessage, executeTool } from '../../tools';
|
||||
import { createLogger } from '../../utils/logger';
|
||||
import { getOpenAIUrl } from '../../utils/api-urls';
|
||||
import { AI_CONFIG } from '../../constants/agent-config';
|
||||
import { escalateToAdmin, shouldEscalate } from '../../services/human-handoff';
|
||||
import type { Env, TelegramUpdate, User, OpenAIAPIResponse, OpenAIMessage } from '../../types';
|
||||
|
||||
const logger = createLogger('message-handler');
|
||||
|
||||
// Register agent singletons with priorities (lower = checked first)
|
||||
const onboardingAgent = new OnboardingAgent();
|
||||
const troubleshootAgent = new TroubleshootAgent();
|
||||
const assetAgent = new AssetAgent();
|
||||
const billingAgent = new BillingAgent();
|
||||
|
||||
registerAgent('onboarding', onboardingAgent, 10);
|
||||
registerAgent('troubleshoot', troubleshootAgent, 20);
|
||||
registerAgent('asset', assetAgent, 30);
|
||||
registerAgent('billing', billingAgent, 40);
|
||||
|
||||
export async function handleMessage(
|
||||
env: Env,
|
||||
update: TelegramUpdate
|
||||
): Promise<void> {
|
||||
if (!update.message?.text) return;
|
||||
|
||||
const message = update.message;
|
||||
const chatId = message.chat.id;
|
||||
const text = message.text!;
|
||||
const telegramUserId = message.from.id.toString();
|
||||
const lang = message.from.language_code ?? 'ko';
|
||||
const requestId = crypto.randomUUID();
|
||||
|
||||
// 1. Check if user is blocked & register/update user
|
||||
const user = await getOrCreateUser(env.DB, telegramUserId, message.from.first_name ?? '', message.from.username);
|
||||
if (!user) {
|
||||
await sendMessage(env.BOT_TOKEN, chatId, getMessage(lang, 'error_general'));
|
||||
return;
|
||||
}
|
||||
if (user.is_blocked) {
|
||||
await sendMessage(env.BOT_TOKEN, chatId, getMessage(lang, 'error_blocked'));
|
||||
return;
|
||||
}
|
||||
|
||||
// 2. Rate limiting
|
||||
if (!(await checkRateLimit(env.RATE_LIMIT_KV, telegramUserId))) {
|
||||
await sendMessage(env.BOT_TOKEN, chatId, getMessage(lang, 'error_rate_limit'));
|
||||
return;
|
||||
}
|
||||
|
||||
// 3. Send typing action
|
||||
await sendChatAction(env.BOT_TOKEN, chatId).catch(() => {});
|
||||
|
||||
try {
|
||||
// 4. Route to active agent session first
|
||||
const agentResponse = await routeToActiveAgent(env.DB, telegramUserId, text, env);
|
||||
if (agentResponse) {
|
||||
const { cleanText, sessionEnded } = cleanSessionMarkers(agentResponse);
|
||||
|
||||
// Handle escalation marker
|
||||
if (agentResponse.includes('__ESCALATE__')) {
|
||||
const cleaned = cleanText.replace('__ESCALATE__', '').trim();
|
||||
await storeConversation(env.DB, user.id, text, cleaned, requestId);
|
||||
await sendMessage(env.BOT_TOKEN, chatId, cleaned);
|
||||
await escalateToAdmin(env, telegramUserId, text, 'agent');
|
||||
return;
|
||||
}
|
||||
|
||||
await storeConversation(env.DB, user.id, text, cleanText, requestId);
|
||||
await sendMessage(env.BOT_TOKEN, chatId, cleanText);
|
||||
|
||||
// Prompt feedback after session end
|
||||
if (sessionEnded) {
|
||||
await promptFeedback(env, chatId, lang, 'agent');
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// 5. Detect intent for new session creation
|
||||
let response: string | null = null;
|
||||
|
||||
if (ONBOARDING_PATTERNS.test(text)) {
|
||||
response = await onboardingAgent.processConsultation(env.DB, telegramUserId, text, env);
|
||||
} else if (TROUBLESHOOT_PATTERNS.test(text)) {
|
||||
response = await troubleshootAgent.processConsultation(env.DB, telegramUserId, text, env);
|
||||
} else if (BILLING_PATTERNS.test(text)) {
|
||||
response = await billingAgent.processConsultation(env.DB, telegramUserId, text, env);
|
||||
} else if (ASSET_PATTERNS.test(text)) {
|
||||
response = await assetAgent.processConsultation(env.DB, telegramUserId, text, env);
|
||||
}
|
||||
|
||||
if (response) {
|
||||
const { cleanText, sessionEnded } = cleanSessionMarkers(response);
|
||||
await storeConversation(env.DB, user.id, text, cleanText, requestId);
|
||||
await sendMessage(env.BOT_TOKEN, chatId, cleanText);
|
||||
if (sessionEnded) {
|
||||
await promptFeedback(env, chatId, lang, 'agent');
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// 6. General AI fallback (no matching agent)
|
||||
const aiResponse = await handleGeneralAI(env, user, text, telegramUserId);
|
||||
await storeConversation(env.DB, user.id, text, aiResponse, requestId);
|
||||
await sendMessage(env.BOT_TOKEN, chatId, aiResponse);
|
||||
|
||||
// 7. Check frustration for potential escalation
|
||||
if (shouldEscalate(text, 0)) {
|
||||
logger.warn('Frustration detected in general AI flow', { telegramUserId });
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error('Message handling error', error as Error, { telegramUserId, requestId });
|
||||
await sendMessage(env.BOT_TOKEN, chatId, getMessage(lang, 'error_general'));
|
||||
}
|
||||
}
|
||||
|
||||
function cleanSessionMarkers(text: string): { cleanText: string; sessionEnded: boolean } {
|
||||
const sessionEnded = text.includes('[세션 종료]') || text.includes('__SESSION_END__');
|
||||
const cleanText = text
|
||||
.replace('[세션 종료]', '')
|
||||
.replace('__SESSION_END__', '')
|
||||
.trim();
|
||||
return { cleanText, sessionEnded };
|
||||
}
|
||||
|
||||
async function getOrCreateUser(
|
||||
db: D1Database,
|
||||
telegramId: string,
|
||||
firstName: string,
|
||||
username?: string
|
||||
): Promise<(User & { id: number }) | null> {
|
||||
try {
|
||||
await db
|
||||
.prepare(
|
||||
`INSERT INTO users (telegram_id, username, first_name, last_active_at)
|
||||
VALUES (?, ?, ?, CURRENT_TIMESTAMP)
|
||||
ON CONFLICT (telegram_id) DO UPDATE SET
|
||||
username = COALESCE(excluded.username, users.username),
|
||||
first_name = COALESCE(excluded.first_name, users.first_name),
|
||||
last_active_at = CURRENT_TIMESTAMP,
|
||||
updated_at = CURRENT_TIMESTAMP`
|
||||
)
|
||||
.bind(telegramId, username ?? null, firstName)
|
||||
.run();
|
||||
|
||||
const user = await db
|
||||
.prepare('SELECT * FROM users WHERE telegram_id = ?')
|
||||
.bind(telegramId)
|
||||
.first<User>();
|
||||
|
||||
return user ?? null;
|
||||
} catch (error) {
|
||||
logger.error('User upsert failed', error as Error, { telegramId });
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async function handleGeneralAI(
|
||||
env: Env,
|
||||
user: User,
|
||||
userMessage: string,
|
||||
telegramUserId: string
|
||||
): Promise<string> {
|
||||
if (!env.OPENAI_API_KEY) {
|
||||
return await handleWorkersAIFallback(env, userMessage);
|
||||
}
|
||||
|
||||
try {
|
||||
// Load recent conversation context
|
||||
const history = await env.DB
|
||||
.prepare(
|
||||
`SELECT role, content FROM conversations
|
||||
WHERE user_id = ? ORDER BY created_at DESC LIMIT ?`
|
||||
)
|
||||
.bind(user.id, user.context_limit)
|
||||
.all<{ role: string; content: string }>();
|
||||
|
||||
const conversationHistory = history.results.reverse();
|
||||
const tools = selectToolsForMessage(userMessage);
|
||||
|
||||
const messages: OpenAIMessage[] = [
|
||||
{
|
||||
role: 'system',
|
||||
content: `당신은 클라우드 호스팅/도메인/서버 관리 서비스의 AI 고객 지원 상담사입니다.
|
||||
한국어로 친절하고 전문적으로 답변하세요.
|
||||
기술 용어는 쉽게 풀어 설명하세요.
|
||||
답변을 모르면 솔직히 모른다고 하고, 관리자 연결을 제안하세요.`,
|
||||
},
|
||||
...conversationHistory.map((m) => ({
|
||||
role: m.role as 'user' | 'assistant',
|
||||
content: m.content,
|
||||
})),
|
||||
{ role: 'user', content: userMessage },
|
||||
];
|
||||
|
||||
const body: Record<string, unknown> = {
|
||||
model: AI_CONFIG.model,
|
||||
messages,
|
||||
max_tokens: 1024,
|
||||
temperature: AI_CONFIG.defaultTemperature,
|
||||
};
|
||||
|
||||
if (tools.length > 0) {
|
||||
body.tools = tools;
|
||||
body.tool_choice = 'auto';
|
||||
}
|
||||
|
||||
const controller = new AbortController();
|
||||
const timeoutId = setTimeout(() => controller.abort(), 25000);
|
||||
|
||||
try {
|
||||
const response = await fetch(getOpenAIUrl(env), {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
Authorization: `Bearer ${env.OPENAI_API_KEY}`,
|
||||
},
|
||||
signal: controller.signal,
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
logger.error('OpenAI API error', new Error(`HTTP ${response.status}`));
|
||||
return await handleWorkersAIFallback(env, userMessage);
|
||||
}
|
||||
|
||||
const data = (await response.json()) as OpenAIAPIResponse;
|
||||
const assistantMessage = data.choices[0].message;
|
||||
|
||||
// Handle tool calls (single round for general AI)
|
||||
if (assistantMessage.tool_calls && assistantMessage.tool_calls.length > 0) {
|
||||
// Add assistant message with tool calls
|
||||
messages.push({
|
||||
role: 'assistant',
|
||||
content: assistantMessage.content,
|
||||
tool_calls: assistantMessage.tool_calls,
|
||||
});
|
||||
|
||||
for (const tc of assistantMessage.tool_calls) {
|
||||
let args: Record<string, unknown>;
|
||||
try {
|
||||
args = JSON.parse(tc.function.arguments) as Record<string, unknown>;
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
const result = await executeTool(tc.function.name, args, env, telegramUserId, env.DB);
|
||||
messages.push({
|
||||
role: 'tool',
|
||||
content: result,
|
||||
tool_call_id: tc.id,
|
||||
name: tc.function.name,
|
||||
});
|
||||
}
|
||||
|
||||
// Second call to synthesize tool results
|
||||
const followUp = await fetch(getOpenAIUrl(env), {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
Authorization: `Bearer ${env.OPENAI_API_KEY}`,
|
||||
},
|
||||
body: JSON.stringify({
|
||||
model: AI_CONFIG.model,
|
||||
messages,
|
||||
max_tokens: 1024,
|
||||
temperature: AI_CONFIG.defaultTemperature,
|
||||
}),
|
||||
});
|
||||
|
||||
if (followUp.ok) {
|
||||
const followUpData = (await followUp.json()) as OpenAIAPIResponse;
|
||||
return followUpData.choices[0].message.content ?? getMessage('ko', 'error_ai_unavailable');
|
||||
}
|
||||
|
||||
// Fallback: return AI text
|
||||
return assistantMessage.content ?? getMessage('ko', 'error_ai_unavailable');
|
||||
}
|
||||
|
||||
return assistantMessage.content ?? getMessage('ko', 'error_ai_unavailable');
|
||||
} finally {
|
||||
clearTimeout(timeoutId);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error('General AI error', error as Error);
|
||||
return await handleWorkersAIFallback(env, userMessage);
|
||||
}
|
||||
}
|
||||
|
||||
async function handleWorkersAIFallback(env: Env, userMessage: string): Promise<string> {
|
||||
try {
|
||||
const result = await env.AI.run('@cf/meta/llama-3.1-8b-instruct-fp8', {
|
||||
messages: [
|
||||
{
|
||||
role: 'system',
|
||||
content: '당신은 클라우드 호스팅 서비스의 AI 고객 지원 상담사입니다. 한국어로 친절하게 답변하세요.',
|
||||
},
|
||||
{ role: 'user', content: userMessage },
|
||||
],
|
||||
});
|
||||
|
||||
if (result && typeof result === 'object' && 'response' in result) {
|
||||
return (result as { response: string }).response;
|
||||
}
|
||||
|
||||
return getMessage('ko', 'error_ai_unavailable');
|
||||
} catch (error) {
|
||||
logger.error('Workers AI fallback error', error as Error);
|
||||
return getMessage('ko', 'error_ai_unavailable');
|
||||
}
|
||||
}
|
||||
|
||||
async function storeConversation(
|
||||
db: D1Database,
|
||||
userId: number,
|
||||
userMessage: string,
|
||||
assistantResponse: string,
|
||||
requestId: string
|
||||
): Promise<void> {
|
||||
try {
|
||||
await db.batch([
|
||||
db.prepare(
|
||||
`INSERT INTO conversations (user_id, role, content, request_id) VALUES (?, 'user', ?, ?)`
|
||||
).bind(userId, userMessage, requestId),
|
||||
db.prepare(
|
||||
`INSERT INTO conversations (user_id, role, content, request_id) VALUES (?, 'assistant', ?, ?)`
|
||||
).bind(userId, assistantResponse, requestId),
|
||||
]);
|
||||
} catch (error) {
|
||||
logger.error('Conversation storage failed', error as Error, { userId, requestId });
|
||||
}
|
||||
}
|
||||
|
||||
async function promptFeedback(
|
||||
env: Env,
|
||||
chatId: number,
|
||||
lang: string,
|
||||
sessionType: string
|
||||
): Promise<void> {
|
||||
try {
|
||||
const text = getMessage(lang, 'feedback_prompt');
|
||||
const keyboard = [
|
||||
[1, 2, 3, 4, 5].map((rating) => ({
|
||||
text: '⭐'.repeat(rating),
|
||||
callback_data: `fb:${sessionType}:${rating}`,
|
||||
})),
|
||||
];
|
||||
await sendMessageWithKeyboard(env.BOT_TOKEN, chatId, text, keyboard);
|
||||
} catch (error) {
|
||||
logger.error('Failed to send feedback prompt', error as Error);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user