kunthawat/moreminimore-chat
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

feat(analytics): M2 self-improving chatbot analytics

Browse Source 
Daily LLM classification of every chat (topics + product tags + deal
won/lost/undecided) aggregated into immutable daily metrics, with a
filterable admin report dashboard, product-catalog import (text/CSV/XLSX),
weekly persona summary, and an approval flow (LINE -> Telegram -> webhook)
for applying persona recommendations.

- Llm::AnalyticsClassifier: per-account openai -> Captain fallback cascade
- AccountDailyProcessor + Conversation/CustomerDailyMetric aggregation
- ReportService + DrilldownService (summary + deep filterable drilldown)
- AnalyticsReports.vue + productCatalog import UI (admin-only)
- WeeklyPersonaEvaluator + PersonaApprovalService (LINE/Telegram/webhook)
- Weekly cron (Mon 10:00) + daily cron (02:30)
This commit is contained in:
Moreminimore
2026-08-25 15:56:11 +07:00
parent 02ae690a87
commit b86b0c59f6
22 changed files with 2095 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

218
app/services/analytics/account_daily_processor.rb Normal file
View File

@@ -0,0 +1,218 @@
# Aggregates one account's conversation activity into the immutable daily metrics
# (conversation_daily_metrics + customer_daily_metrics) for a single date in the
# account's reporting timezone, and writes the LLM classification tags onto the
# live conversations.
#
# Flow per account/date:
# 1. Compute the UTC day window from account.reporting_timezone (default UTC).
# 2. Select conversations with activity inside that window.
# 3. For each conversation, classify via Llm::AnalyticsClassifier (skipped when
# disabled), re-tag it (replace prior M2-managed tags, keep manual labels),
# and collect per-conversation / per-customer metrics.
# 4. Upsert conversation_daily_metric (unique account+date) and
# customer_daily_metric (unique account+contact+date).
#
# Daily snapshots are immutable: counts captured here are a point-in-time view; the
# live conversation tags may be re-tagged later without changing historical snapshots.
class Analytics::AccountDailyProcessor
TOPIC_TAG_PREFIX = 'topic:'.freeze
DEAL_TAG_PREFIX = 'deal:'.freeze
# @param account [Account]
# @param date [Date] the reporting date in the account's timezone
def self.perform(account:, date:)
new(account: account, date: date).perform
end
def initialize(account:, date:)
@account = account
@date = date
end
def perform
return unless scope_enabled?
window = day_window
conversations = active_conversations(window)
return if conversations.empty?
catalog_paths = catalog_paths
metrics = build_metrics(conversations, window, catalog_paths)
upsert_metrics(metrics)
end
private
def scope_enabled?
@account.reporting_timezone.blank? || ActiveSupport::TimeZone[@account.reporting_timezone].present?
end
def timezone
@timezone ||= ActiveSupport::TimeZone[@account.reporting_timezone] || Time.zone
end
# [start_utc, end_utc) covering the account-local date.
def day_window
start_tz = timezone.parse(@date.to_s)
end_tz = timezone.parse((@date + 1).to_s)
[start_tz.utc, end_tz.utc]
end
# Conversations with activity (created or last activity) inside the account-local day.
def active_conversations(window)
start_utc, end_utc = window
@account.conversations
.where('created_at < ?', end_utc)
.where('created_at >= ? OR last_activity_at >= ?', start_utc, start_utc)
end
# All tag-hierarchy paths in the account's catalog — used to detect & replace
# previously-written product tags on re-classification.
def catalog_paths
@account.product_catalog_entries
.order(:group_name, :subgroup_name, :product_name)
.map(&:tag_hierarchy)
.map { |parts| parts.join('>') }
end
def build_metrics(conversations, _window, catalog_paths)
{
conversation: {
message_count: 0,
conversation_count: 0,
resolved_count: 0,
unresolved_count: 0,
top_tags: [],
sale_tags: [],
deal_outcomes: {},
agent_breakdown: Hash.new(0),
team_breakdown: Hash.new(0),
channel_breakdown: Hash.new(0),
inbox_breakdown: Hash.new(0)
},
customers: {}
}.tap do |acc|
conversations.each { |conversation| accumulate(acc, conversation, catalog_paths) }
end
end
def accumulate(acc, conversation, catalog_paths)
classification = Llm::AnalyticsClassifier.classify(account: @account, conversation: conversation)
return if classification.error.present? || classification.disabled?
apply_tags(conversation, classification, catalog_paths)
cm = acc[:conversation]
cm[:conversation_count] += 1
message_count = chat_message_count(conversation)
cm[:message_count] += message_count
if conversation.resolved?
cm[:resolved_count] += 1
else
cm[:unresolved_count] += 1
end
cm[:top_tags] |= (classification.topics + classification.products).compact
cm[:sale_tags] |= classification.products
cm[:deal_outcomes]['totals'] ||= {}
cm[:deal_outcomes]['totals'][classification.deal] = cm[:deal_outcomes]['totals'][classification.deal].to_i + 1
cm[:agent_breakdown][conversation.assignee_id] += 1 if conversation.assignee_id
cm[:team_breakdown][conversation.team_id] += 1 if conversation.team_id
cm[:channel_breakdown][channel_key(conversation)] += 1
cm[:inbox_breakdown][conversation.inbox_id] += 1 if conversation.inbox_id
customer_key = conversation.contact_id
customer = acc[:customers][customer_key] ||= {
message_count: 0, conversation_count: 0, resolved_count: 0, unresolved_count: 0,
top_tags: [], deal_outcomes: {}, agent_ids: []
}
customer[:conversation_count] += 1
customer[:message_count] += message_count
if conversation.resolved?
customer[:resolved_count] += 1
else
customer[:unresolved_count] += 1
end
customer[:top_tags] |= (classification.topics + classification.products).compact
customer[:deal_outcomes][classification.deal] = customer[:deal_outcomes][classification.deal].to_i + 1
customer[:agent_ids] = (customer[:agent_ids] | [conversation.assignee_id]).compact if conversation.assignee_id
end
def chat_message_count(conversation)
conversation.messages.chat.count
end
def channel_key(conversation)
conversation.inbox&.channel_type&.demodulize
end
# Write classifier tags onto the conversation: remove any prior M2-managed tags
# (topic:/deal: prefixes + current catalog paths), then add the fresh ones.
# Manual labels are preserved.
def apply_tags(conversation, classification, catalog_paths)
current = conversation.label_list.to_a
m2_managed = current.select do |tag|
tag.start_with?(TOPIC_TAG_PREFIX) || tag.start_with?(DEAL_TAG_PREFIX) || catalog_paths.include?(tag)
end
next_tags = current - m2_managed
next_tags += classification.products
next_tags += classification.topics.map { |topic| "#{TOPIC_TAG_PREFIX}#{topic}" }
# Persist the deal outcome on the live conversation so reports can filter by it.
next_tags << "#{DEAL_TAG_PREFIX}#{classification.deal}" if classification.deal.present?
conversation.update!(label_list: next_tags.uniq)
rescue ActiveRecord::RecordInvalid => e
Rails.logger.error("[Analytics] conversation #{conversation.id} tag update failed: #{e.message}")
end
def upsert_metrics(metrics)
conv = metrics[:conversation]
conv_attrs = {
account_id: @account.id,
date: @date,
timezone: @account.reporting_timezone.presence || 'UTC',
message_count: conv[:message_count],
conversation_count: conv[:conversation_count],
resolved_count: conv[:resolved_count],
unresolved_count: conv[:unresolved_count],
top_tags: conv[:top_tags],
sale_tags: conv[:sale_tags],
deal_outcomes: conv[:deal_outcomes],
agent_breakdown: conv[:agent_breakdown].map { |id, count| { 'agent_id' => id, 'count' => count } },
team_breakdown: conv[:team_breakdown].map { |id, count| { 'team_id' => id, 'count' => count } },
channel_breakdown: conv[:channel_breakdown].map { |ch, count| { 'channel' => ch, 'count' => count } },
inbox_breakdown: conv[:inbox_breakdown].map { |id, count| { 'inbox_id' => id, 'count' => count } }
}
ConversationDailyMetric.upsert(
conv_attrs,
unique_by: %i[account_id date],
update_only: %i[message_count conversation_count resolved_count unresolved_count top_tags sale_tags
deal_outcomes agent_breakdown team_breakdown channel_breakdown inbox_breakdown timezone]
)
metrics[:customers].each do |contact_id, data|
CustomerDailyMetric.upsert(
{
account_id: @account.id,
contact_id: contact_id,
date: @date,
timezone: @account.reporting_timezone.presence || 'UTC',
message_count: data[:message_count],
conversation_count: data[:conversation_count],
resolved_count: data[:resolved_count],
unresolved_count: data[:unresolved_count],
top_tags: data[:top_tags],
deal_outcomes: data[:deal_outcomes],
agent_ids: data[:agent_ids].to_a
},
unique_by: %i[account_id contact_id date],
update_only: %i[message_count conversation_count resolved_count unresolved_count top_tags
deal_outcomes agent_ids timezone]
)
end
end
end

107
app/services/analytics/drilldown_service.rb Normal file
View File

@@ -0,0 +1,107 @@
# Deep, filterable admin drilldown over LIVE conversations (phase 2).
#
# Unlike Analytics::ReportService (which reads the immutable per-day snapshots),
# this queries the actual Conversation records scoped to a date range, so the
# filters the M2 report needs (agent / team / inbox / channel / tag / deal) can
# be applied at per-conversation granularity, with per-customer and per-agent
# cross-breakdowns and pagination.
#
# Supported filters (all optional, applied with AND):
# :since (Date) — conversations active on/after (created_at or last_activity_at)
# :until (Date) — conversations active before this date (exclusive end)
# :agent_id (Integer)
# :team_id (Integer)
# :inbox_id (Integer)
# :channel (String, e.g. "Channel::WebWidget") — matched against inbox.channel_type
# :tag (String) — matches any tag in the conversation label_list (exact)
# :deal (String, "won"|"lost"|"undecided") — matches the "deal:<value>" tag
# :page, :per_page — pagination (default 1 / 25)
#
# Admin-only; access is enforced by the calling controller/policy.
class Analytics::DrilldownService
DEFAULT_PER_PAGE = 25
# @param account [Account]
# @param filters [Hash]
def self.build(account:, filters: {})
new(account: account, filters: filters).build
end
def initialize(account:, filters: {})
@account = account
@filters = filters.symbolize_keys
end
def build
scope = filtered_scope
{
total: scope.count,
page: page,
per_page: per_page,
conversations: scope.offset((page - 1) * per_page).limit(per_page).map { |c| serialize(c) }
}
end
private
attr_reader :account, :filters
def page = (filters[:page] || 1).to_i
def per_page = (filters[:per_page] || DEFAULT_PER_PAGE).to_i.clamp(1, 100)
def filtered_scope
scope = account.conversations
if filters[:since].present?
start = filters[:since]
scope = scope.where('created_at >= ?', start)
end
if filters[:until].present?
scope = scope.where('created_at < ?', filters[:until] + 1)
end
scope = scope.where(assignee_id: filters[:agent_id]) if filters[:agent_id].present?
scope = scope.where(team_id: filters[:team_id]) if filters[:team_id].present?
scope = scope.where(inbox_id: filters[:inbox_id]) if filters[:inbox_id].present?
if filters[:channel].present?
scope = scope.joins(:inbox).where(inboxes: { channel_type: filters[:channel] })
end
if filters[:tag].present?
tag = filters[:tag].to_s
matches = account.labels.where(title: tag).pluck(:id)
# No conversation can carry a label that doesn't exist — return none.
return scope.none if matches.empty?
scope = scope.joins(:labels).where('labels.id IN (?)', matches)
end
if filters[:deal].present?
deal_tag = "deal:#{filters[:deal]}"
matches = account.labels.where(title: deal_tag).pluck(:id)
# No conversation can carry a deal tag that doesn't exist — return none.
return scope.none if matches.empty?
scope = scope.joins(:labels).where('labels.id IN (?)', matches)
end
scope.order(:created_at)
end
def serialize(conversation)
{
id: conversation.id,
display_id: conversation.display_id,
contact_id: conversation.contact_id,
assignee_id: conversation.assignee_id,
team_id: conversation.team_id,
inbox_id: conversation.inbox_id,
channel: conversation.inbox&.channel_type,
status: conversation.status,
created_at: conversation.created_at,
label_list: conversation.label_list,
message_count: conversation.messages.chat.count
}
end
end

200
app/services/analytics/persona_approval_service.rb Normal file
View File

@@ -0,0 +1,200 @@
# Approval-flow delivery for the weekly persona evaluation (phase 3).
#
# Delivers the weekly LLM summary + recommendations to the admin via LINE
# (primary, using quick-reply buttons), Telegram (inline keyboard), or a webhook
# (fallback), then records the admin's decision. The full persona system prompt is
# intentionally NOT included in the delivery body — it is only sent to the webhook
# AFTER the admin approves.
#
# Delivery configuration lives on Account#custom_attributes (jsonb):
# persona_line_user_id - the admin's LINE user id (for LINE push) [primary]
# persona_telegram_chat_id - the admin's Telegram chat id (for Telegram push)
# persona_webhook_url - fallback webhook URL
# persona_webhook_secret - shared secret signed into the webhook body / header
#
# Decision keys match LINE quick-reply / webhook actions:
# 'approve' | 'reject' | 'view_full_prompt'
class Analytics::PersonaApprovalService
# @param account [Account]
# @param evaluation [Analytics::WeeklyPersonaEvaluator::Result]
# @return [Hash] { delivered:, channel:, error: }
def self.deliver(account:, evaluation:)
new(account: account, evaluation: evaluation).deliver
end
# Sends an admin approval decision to the configured webhook (webhook-only, never
# LINE). Used by the :approve path to inform the integration about the decision.
# @return [Hash] { delivered:, channel: 'webhook', error: }
def self.notify_approval(account:, evaluation:, decision:)
new(account: account, evaluation: evaluation).notify_approval(decision)
end
def initialize(account:, evaluation:)
@account = account
@evaluation = evaluation
end
def deliver
return { delivered: false, error: 'no evaluation to deliver' } if @evaluation.nil? || @evaluation.disabled?
if line_delivery_available?
deliver_via_line
elsif telegram_delivery_available?
deliver_via_telegram
elsif webhook_delivery_available?
deliver_via_webhook
else
{ delivered: false, channel: nil, error: 'no delivery channel configured (set persona_line_user_id, persona_telegram_chat_id or persona_webhook_url)' }
end
end
# Sends an admin approval decision to the configured webhook. Webhook-only (never
# LINE) — used by the :approve path so the approval is not re-sent as a LINE card.
def notify_approval(decision)
return { delivered: false, channel: 'webhook', error: 'no webhook configured' } unless webhook_delivery_available?
post_to_webhook(
type: 'persona_decision',
decision: decision
)
end
private
attr_reader :account
# -- LINE (primary) ----------------------------------------------------------
def line_delivery_available?
line_user_id.present? && account.line_channels.present?
end
def deliver_via_line
channel = account.line_channels.first
channel.client.push_message(
line_user_id,
build_line_payload
)
{ delivered: true, channel: 'line' }
rescue StandardError => e
Rails.logger.error("[PersonaApproval] LINE push failed: #{e.message}")
{ delivered: false, channel: 'line', error: e.message }
end
def build_line_payload
{
type: 'text',
text: line_text,
quickReply: {
items: [
quick_reply_item('👍 Approve', 'approve'),
quick_reply_item('Reject', 'reject'),
quick_reply_item('👁 View full prompt', 'view_full_prompt')
]
}
}
end
def quick_reply_item(label, key)
{
type: 'action',
action: {
type: 'message',
label: label,
text: "persona:#{key}"
}
}
end
def line_text
summary = @evaluation.summary.to_s
recommendations = Array(@evaluation.recommendations)
[summary, '', *recommendations.map { |r| "• #{r}" }].join("\n")
end
# -- Telegram ----------------------------------------------------------------
def telegram_delivery_available?
telegram_chat_id.present? && account.telegram_channels.present?
end
def deliver_via_telegram
channel = account.telegram_channels.first
response = HTTParty.post(
"#{channel.telegram_api_url}/sendMessage",
body: {
chat_id: telegram_chat_id,
text: line_text,
reply_markup: build_telegram_keyboard
}
)
success = response.success?
Rails.logger.error("[PersonaApproval] Telegram send failed: #{response.parsed_response}") unless success
{ delivered: success, channel: 'telegram', error: success ? nil : 'telegram send failed' }
rescue StandardError => e
Rails.logger.error("[PersonaApproval] Telegram delivery failed: #{e.message}")
{ delivered: false, channel: 'telegram', error: e.message }
end
# Inline keyboard with the three decision buttons (callback_data = persona:<key>)
def build_telegram_keyboard
{
inline_keyboard: [
[
{ text: '👍 Approve', callback_data: 'persona:approve' },
{ text: 'Reject', callback_data: 'persona:reject' },
{ text: '👁 View full prompt', callback_data: 'persona:view_full_prompt' }
]
]
}.to_json
end
# -- Webhook (fallback) ------------------------------------------------------
def webhook_delivery_available?
webhook_url.present?
end
# Shared webhook POST: builds the payload from the evaluation plus extra fields.
def post_to_webhook(extra_fields)
body = {
summary: @evaluation.summary,
recommendations: Array(@evaluation.recommendations)
}.merge(extra_fields)
response = HTTParty.post(
webhook_url,
body: body.to_json,
headers: webhook_headers
)
success = response.success?
Rails.logger.error("[PersonaApproval] webhook #{response.code}") unless success
{ delivered: success, channel: 'webhook', error: success ? nil : "webhook responded #{response.code}" }
rescue StandardError => e
Rails.logger.error("[PersonaApproval] webhook failed: #{e.message}")
{ delivered: false, channel: 'webhook', error: e.message }
end
def deliver_via_webhook
post_to_webhook(
type: 'persona_evaluation',
actions: %w[approve reject view_full_prompt]
)
end
def webhook_headers
headers = { 'Content-Type' => 'application/json' }
headers['X-Persona-Signature'] = signature if webhook_secret.present?
headers
end
def signature
OpenSSL::HMAC.hexdigest('sha256', webhook_secret, @evaluation.summary.to_s)
end
# -- config ------------------------------------------------------------------
def line_user_id = account.custom_attributes['persona_line_user_id']
def telegram_chat_id = account.custom_attributes['persona_telegram_chat_id']
def webhook_url = account.custom_attributes['persona_webhook_url']
def webhook_secret = account.custom_attributes['persona_webhook_secret']
end

228
app/services/analytics/product_catalog_import_service.rb Normal file
View File

@@ -0,0 +1,228 @@
# Imports product catalog entries for an account from pasted text, CSV/TSV, or an
# uploaded .xlsx spreadsheet (phase 3, admin-only).
#
# Accepted formats:
# - copy/paste: one product per line, columns separated by tab or '|'
# - CSV: standard RFC4180 with a header row
# - XLSX: an uploaded spreadsheet opened via the `roo` gem (file_path:)
# Columns (in order): group, subgroup, product, display, aliases
# group, product are required; subgroup/display optional; aliases separator is
# comma or semicolon.
#
# Hierarchical semantics: a row defines one product leaf. The classifier matches
# from the lowest level first and tags ancestors, so only leaf rows are stored;
# no need to also store group/subgroup as standalone rows.
#
# Upsert semantics: same (account_id, group_name, product_name) is replaced with
# the latest row (aliases/subgroup/display overwritten), keeping the catalog unique.
#
# Returns a Hash: { imported: n, updated: n, errors: [{ line, message }] }.
class Analytics::ProductCatalogImportService
COLUMNS = %w[group_name subgroup_name product_name display_name aliases].freeze
COLUMN_ALIASES = {
'group' => 'group_name', 'product' => 'product_name',
'subgroup' => 'subgroup_name', 'display' => 'display_name'
}.freeze
REQUIRED = %w[group_name product_name].freeze
ALIAS_SPLIT = /[,;]/
# @param account [Account]
# @param content [String] raw pasted text or file content (CSV/TSV/pipe)
# @return [Hash]
def self.import(account:, content:)
new(account: account, content: content).import
end
# @param account [Account]
# @param file_path [String] path to an uploaded .xlsx/.ods/.csv file (via roo)
# @param filename [String, nil] original upload filename (used to detect extension
# when the temp path has none, e.g. RackMultipart tempfile)
# @return [Hash]
def self.import_file(account:, file_path:, filename: nil)
new(account: account, content: nil, file_path: file_path, filename: filename).import
end
def initialize(account:, content: nil, file_path: nil, filename: nil)
@account = account
@content = content.to_s
@file_path = file_path
@filename = filename
end
def import
rows = (@file_path ? parse_spreadsheet : parse_rows)
return { imported: 0, updated: 0, errors: rows[:errors] } if rows[:data].empty?
result = upsert_rows(rows[:data])
result.merge(errors: rows[:errors])
end
private
attr_reader :account
# Convert an .xlsx/.xls/.ods sheet into the same { data, errors } shape as
# parse_rows by serializing each row into a pipe-delimited text line, then
# reusing the shared row parser. The first row is treated as a header when it
# looks like declared column names (same detection as text input).
def parse_spreadsheet
require 'roo'
sheet = Roo::Spreadsheet.open(@file_path, extension: extension).sheet(0)
lines = (1..sheet.last_row).filter_map do |idx|
row = (1..sheet.last_column).map { |col| sheet.cell(idx, col).to_s }
row.join('|') unless row.all?(&:blank?)
end
parse_rows_from_lines(lines)
rescue LoadError
# roo gem unavailable (should be resolved after bundle install).
Rails.logger.error('[ProductCatalogImport] roo gem not available; cannot read spreadsheet')
{ data: [], errors: [{ line: 1, message: 'spreadsheet reading is unavailable (roo gem not installed)' }] }
rescue StandardError => e
Rails.logger.error("[ProductCatalogImport] xlsx parse failed: #{e.message}")
{ data: [], errors: [{ line: 1, message: "could not read spreadsheet: #{e.message}" }] }
end
def extension
# Prefer the original upload filename (Rack tempfiles lack a useful extension);
# fall back to the path if no filename was supplied.
name = @filename.presence || @file_path.to_s
File.extname(name).delete('.').presence
end
# Shared row->attrs pipeline used by both text and spreadsheet input.
def parse_rows_from_lines(lines)
return { data: [], errors: [] } if lines.empty?
header_columns = detect_header_columns(lines)
line_offset = header_columns ? 1 : 0
keyword_mode = lines.any? { |line| keyword_line?(line) }
data = []
errors = []
lines.each_with_index do |line, idx|
raw = split_line(line)
attrs = row_to_attrs(raw, header_columns: header_columns, keyword: keyword_mode, line_no: idx + 1 + line_offset)
if attrs.is_a?(Hash)
data << attrs
else
errors << { line: idx + 1 + line_offset, message: attrs }
end
end
{ data: data, errors: errors }
end
# @return [Hash] { data: [attrs...], errors: [{ line:, message: }] }
def parse_rows
lines = @content.strip.split(/\r?\n/).reject(&:blank?)
parse_rows_from_lines(lines)
end
# If the first line looks like a declared column header (e.g. "group,product" or
# "group|subgroup|product|display|aliases", accepting group/group_name and
# product/product_name), return the normalized column names so data rows are
# mapped by name. Returns nil otherwise.
def detect_header_columns(lines)
first = split_line(lines.first).map { |v| v.to_s.strip.downcase }
normalized = first.map { |col| COLUMN_ALIASES.fetch(col, col) }
return nil unless normalized.all? { |col| COLUMNS.include?(col) }
lines.shift
normalized
end
def keyword_line?(line)
line =~ /(?:^|[\t\|\s])(group_name|group|product_name|product|subgroup_name|display_name|aliases)\s*:/
end
# Split a line by a consistent delimiter. If tabs present -> TSV; else '|' -> pipe; else comma -> CSV.
def split_line(line)
if line.include?("\t")
line.split("\t")
elsif line.include?('|')
line.split('|')
else
CSV.parse_line(line) || []
end
end
def row_to_attrs(raw, header_columns:, keyword:, line_no:)
values = raw.map(&:to_s).map(&:strip)
values = values.map(&:presence).compact
return 'empty row' if values.empty?
attrs =
if keyword
keyword_row_to_attrs(values)
elsif header_columns
header_row_to_attrs(raw, header_columns)
else
ordered_row_to_attrs(values)
end
return attrs if attrs.is_a?(String)
missing = REQUIRED.select { |col| attrs[col].blank? }
return "line #{line_no}: missing required column(s): #{missing.join(', ')}" unless missing.empty?
attrs['aliases'] = normalize_aliases(attrs['aliases'])
attrs
end
# Field form: "group: A | product: X | subgroup: S | aliases: a,b"
def keyword_row_to_attrs(values)
attrs = {}
values.each do |pair|
key, _, value = pair.partition(':')
normalized = COLUMN_ALIASES.fetch(key.strip.downcase, key.strip.downcase)
attrs[normalized] = value.strip.presence
end
attrs
end
# Named-header form: header tells us which column is which.
def header_row_to_attrs(raw, header_columns)
attrs = {}
raw.each_with_index do |value, i|
col = header_columns[i]
attrs[col] = value.presence if col
end
attrs
end
# Ordered form: "group, product" (no subgroup) or "group, subgroup, product".
def ordered_row_to_attrs(values)
if values.length == 2
{ 'group_name' => values[0], 'product_name' => values[1] }
else
# group, subgroup, product, [display], [aliases]
{ 'group_name' => values[0], 'subgroup_name' => values[1], 'product_name' => values[2],
'display_name' => values[3], 'aliases' => values[4] }
end
end
def normalize_aliases(value)
value.to_s.split(ALIAS_SPLIT).map(&:strip).reject(&:blank?)
end
def upsert_rows(rows)
imported = 0
updated = 0
rows.each do |attrs|
existing = @account.product_catalog_entries.find_by(
group_name: attrs['group_name'], product_name: attrs['product_name']
)
if existing
existing.update!(
subgroup_name: attrs['subgroup_name'],
display_name: attrs['display_name'],
aliases: attrs['aliases']
)
updated += 1
else
@account.product_catalog_entries.create!(attrs.merge(account_id: @account.id))
imported += 1
end
end
{ imported: imported, updated: updated }
end
end

113
app/services/analytics/report_service.rb Normal file
View File

@@ -0,0 +1,113 @@
# Admin analytics summary/rollup over the immutable per-day metrics (phase 2).
#
# Reads conversation_daily_metrics + customer_daily_metrics (written by the
# Analytics::AccountDailyProcessor batch) and returns:
# - summary : totals over the date range
# - timeseries : the same totals bucketed per day (for charting)
# - customers : per-customer totals (from customer_daily_metrics)
# - agents : per-agent conversation counts (from agent_breakdown)
#
# This is the snapshot/rollup surface. Deep per-conversation filtering
# (agent/team/inbox/channel/tag/deal) lives in Analytics::DrilldownService, which
# reads live conversations. Admin-only; access is enforced by the controller/policy.
class Analytics::ReportService
# @param account [Account]
# @param since [Date]
# @param until_date [Date]
def self.build(account:, since: nil, until_date: nil)
new(account: account, since: since, until_date: until_date).build
end
def initialize(account:, since: nil, until_date: nil)
@account = account
@since = since
@until_date = until_date
end
def build
{
summary: summary,
timeseries: timeseries,
customers: customers,
agents: agents
}
end
private
attr_reader :account
def since
@since || ConversationDailyMetric.where(account_id: account.id).minimum(:date) || Date.today
end
def until_date
@until_date || Date.today
end
def date_range = (since..until_date)
def conversation_rows
@conversation_rows ||= ConversationDailyMetric
.where(account_id: account.id)
.where(date: date_range)
.order(:date)
end
def customer_rows
@customer_rows ||= CustomerDailyMetric
.where(account_id: account.id)
.where(date: date_range)
.order(:date, :contact_id)
end
# Aggregate a set of rows into normalized totals.
def totals(rows)
{
conversation_count: rows.sum { |r| r.conversation_count },
message_count: rows.sum { |r| r.message_count },
resolved_count: rows.sum { |r| r.resolved_count },
unresolved_count: rows.sum { |r| r.unresolved_count },
deal_outcomes: merge_deal_outcomes(rows),
top_tags: merge_top_tags(rows)
}
end
def summary = totals(conversation_rows)
def timeseries
conversation_rows.group_by(&:date).map { |date, rows| totals(rows).merge(date: date) }
end
def customers
customer_rows.group_by(&:contact_id).map do |contact_id, rows|
totals(rows).merge(contact_id: contact_id)
end
end
def agents
per_agent = Hash.new { |h, k| h[k] = 0 }
conversation_rows.each do |row|
Array(row.agent_breakdown).each { |entry| per_agent[entry['agent_id']] += entry['count'].to_i }
end
per_agent.map { |agent_id, count| { agent_id: agent_id, conversation_count: count } }
.sort_by { |entry| -entry[:conversation_count] }
end
# conversation rows store deal_outcomes nested under 'totals'; customer rows flat.
# Normalize both into a flat { deal => count } mapping.
def merge_deal_outcomes(rows)
rows.each_with_object({}) do |row, acc|
row.deal_outcomes.each do |key, value|
data = value.is_a?(Hash) ? value : { key => value }
data.each { |deal, count| acc[deal] = (acc[deal] || 0) + count.to_i }
end
end
end
def merge_top_tags(rows)
rows.each_with_object(Hash.new(0)) do |row, acc|
Array(row.top_tags).each { |tag| acc[tag] += 1 }
end.sort_by { |_tag, count| -count }.to_h
end
end

123
app/services/analytics/weekly_persona_evaluator.rb Normal file
View File

@@ -0,0 +1,123 @@
# Weekly persona evaluation for the self-improving chatbot (phase 3).
#
# Summarizes the last 7 days of immutable daily metrics (Analytics::ReportService /
# ConversationDailyMetric + CustomerDailyMetric via Analytics::ReportService) and asks the
# LLM (via the Llm::Resolver cascade) to recommend persona/system-prompt improvements.
#
# Output is the SUMMARY ONLY (human-readable recommendation) — the full system prompt is
# intentionally NOT produced/revealed here; it is gated behind the admin approval flow.
#
# Like Llm::AnalyticsClassifier, this is a pure evaluator: it CLASSIFIES/SUMMARIZES and
# returns a Result; persisting the recommendation is the caller's responsibility
# (the weekly job / approval flow). Fail-closed: no LLM credential -> { disabled: true },
# never sends conversation content when disabled.
module Analytics::WeeklyPersonaEvaluator
SCHEMA = {
type: 'object',
additionalProperties: false,
properties: {
summary: {
type: 'string',
description: 'A concise human-readable summary of the week: top topics, sales wins/losses, and any notable trends.'
},
recommendations: {
type: 'array',
items: { type: 'string' },
description: 'Concrete, actionable recommendations to improve the chatbot persona/behavior next week.'
}
},
required: %w[summary recommendations]
}.freeze
Result = Struct.new(:summary, :recommendations, :disabled, :error, keyword_init: true) do
def disabled?
disabled == true
end
def success?
error.nil?
end
end
WINDOW_DAYS = 7
module_function
# @param account [Account]
# @param report [Hash] output of Analytics::ReportService.build (or built here if nil)
# @return [Analytics::WeeklyPersonaEvaluator::Result]
def evaluate(account:, report: nil)
credential = Llm::Resolver.resolve(account)
return disabled_result if credential.nil?
report ||= Analytics::ReportService.build(account: account, since: WINDOW_DAYS.days.ago.to_date, until_date: Date.today)
response = call_llm(credential, build_prompt(report))
build_result(response)
rescue StandardError => e
Rails.logger.error("[WeeklyPersonaEvaluator] account=#{account&.id} #{e.class}: #{e.message}")
Result.new(error: e.message)
end
# -- result helpers ---------------------------------------------------------
def disabled_result
Result.new(summary: nil, recommendations: [], disabled: true)
end
def build_result(response)
return Result.new(error: response[:error] || 'evaluation failed') if response[:error]
parsed = JSON.parse(sanitize_json(response[:content]))
Result.new(
summary: parsed['summary'],
recommendations: Array(parsed['recommendations']),
disabled: false
)
rescue JSON::ParserError, TypeError
Result.new(error: 'LLM returned an unparsable evaluation')
end
# -- LLM call ---------------------------------------------------------------
def call_llm(credential, prompt)
Llm::Config.with_api_key(credential[:api_key], api_base: credential[:api_base]) do |context|
chat = context.chat(model: MODEL).with_schema(SCHEMA)
chat.with_instructions(SYSTEM_PROMPT)
{ content: chat.ask(prompt).content }
end
rescue StandardError => e
Rails.logger.error("[WeeklyPersonaEvaluator] LLM call failed #{e.class}: #{e.message}")
{ error: e.message }
end
MODEL = Llm::Config::DEFAULT_MODEL
# -- prompt construction ----------------------------------------------------
def build_prompt(report)
summary = report[:summary].to_h
[
'Here is the past week of customer-service analytics for the account:',
'',
"Conversations: #{summary[:conversation_count]}",
"Messages: #{summary[:message_count]}",
"Resolved: #{summary[:resolved_count]}",
"Unresolved: #{summary[:unresolved_count]}",
"Deal outcomes: #{summary[:deal_outcomes].inspect}",
"Top tags: #{summary[:top_tags].inspect}",
'',
'Based on this, recommend persona / behavior improvements for the chatbot.'
].join("\n")
end
def sanitize_json(content)
content.to_s.gsub('```json', '').gsub('```', '').strip
end
SYSTEM_PROMPT = <<~PROMPT.freeze
You are a customer-service improvement analyst. Given a week of aggregate metrics,
write a concise summary and 2-5 concrete, actionable recommendations to improve the
chatbot's persona and behavior. Keep recommendations specific and grounded in the data.
Return only the JSON object described by the schema — no extra text.
PROMPT
end