Promtail、Loki 与 Grafana 可以完成日志采集、存储、检索和可视化，但真正发生故障时，值班人员面对的往往不是“缺少日志”，而是信息过载：同一异常在多个实例重复出现，堆栈被截断，请求标识散落在不同日志流中，告警消息又只包含一条匹配记录。
把大模型接到告警链路，可以辅助完成错误聚类、时间线整理和排查建议生成。不过，直接把原始日志发送给外部接口会引入新的问题：日志可能含有令牌、手机号或业务数据；模型可能把推测写成事实；超长上下文会增加延迟和费用；告警重试还可能造成重复调用。
因此，合理目标不是让模型“自动判定根因”，而是建立一条受控的摘要流水线：Loki 负责确定性筛选，服务端负责去重、脱敏和裁剪，模型只处理经过约束的上下文，最终结果保留证据引用并交给人工确认。
整体链路与职责边界
建议将链路拆成五层：
- Promtail 在采集侧解析 JSON 日志，并移除明确禁止外发的字段。
- Loki 保存日志，通过 LogQL 规则检测错误率或特定事件。
- Alertmanager 完成分组、抑制和 Webhook 路由。
- 摘要服务根据告警标签回查限定时间窗内的日志，执行二次脱敏、去重和长度控制。
- 模型输出结构化摘要，摘要服务校验格式后再写入工单或通知渠道。
这里有一个重要边界：标签适合放低基数、可枚举的维度，例如服务名、环境和日志级别；request_id、用户 ID、完整 URL 等高基数字段应留在日志正文中。把高基数字段提升为 Loki 标签会显著增加流数量和索引压力，实际影响取决于日志规模、字段分布及 Loki 部署方式。
第一步：采集时完成结构化解析
假设应用输出 JSON 日志，可以使用如下 Promtail 配置。示例只展示关键部分，文件路径、租户认证和 Loki 地址应按部署环境调整：
server:
http_listen_port: 9080
grpc_listen_port: 0
positions:
filename: /var/lib/promtail/positions.yaml
clients:
- url: ${
LOKI_PUSH_URL}
scrape_configs:
- job_name: application
static_configs:
- targets: [localhost]
labels:
job: application
__path__: /var/log/apps/*.json
pipeline_stages:
- json:
expressions:
level: level
service: service
message: message
request_id: request_id
authorization: authorization
- labels:
level:
service:
- template:
source: authorization
template: "[REDACTED]"
- output:
source: message
采集侧脱敏只能作为第一道防线。上述配置改变了最终发送的日志正文，并不自动覆盖所有嵌套字段、异常堆栈或其他文件副本。生产环境应先检查应用实际日志格式，再决定是删除字段、替换字段，还是要求应用从源头禁止记录凭据。
如果仍在使用 Promtail，还应确认当前维护状态和迁移计划；具体支持周期应以 Grafana 官方当前文档为准。采集端也可以替换为支持向 Loki 写入的其他代理，但配置语法不会完全相同。
第二步：用 LogQL 产生有意义的告警
不要为每条 error 日志单独发送告警。先在固定窗口内聚合，再由 Alertmanager 按服务和环境分组。以下规则统计五分钟内各服务的错误日志数量：
groups:
- name: application-log-alerts
interval: 1m
rules:
- alert: ApplicationErrorBurst
expr: |
sum by (service, env) (
count_over_time({job="application", level="error"}[5m])
) > 20
for: 2m
labels:
severity: warning
annotations:
summary: "服务 {
{ $labels.service }} 出现持续错误日志"
query: '{job="application",service="{
{ $labels.service }}",level="error"}'
阈值 20 只是配置示例，不代表适合任何生产系统。应根据历史基线、发布时段和业务容忍度设置，并观察误报与漏报。对于流量波动明显的服务，可以用错误量与请求量的比率告警，但前提是请求总量能够从日志或指标中可靠取得。
Alertmanager 中可按 alertname、service 和 env 分组，并将同组告警批量发送给摘要服务：
route:
receiver: log-summary-webhook
group_by: [alertname, service, env]
group_wait: 30s
group_interval: 5m
repeat_interval: 2h
receivers:
- name: log-summary-webhook
webhook_configs:
- url: http://log-summarizer:8080/alerts
send_resolved: true
摘要服务必须验证请求来源。可行方式包括仅开放内网地址、在反向代理层启用双向 TLS，或由双方约定签名头。具体选择取决于现有基础设施，不能仅依赖 URL 难以猜测。
第三步：限制回查范围并执行二次脱敏
Webhook 不应把收到的注解直接拼成提示词。服务端应使用自己维护的 LogQL 模板，根据经过白名单校验的 service、env 和时间范围查询 Loki，避免标签值变成查询注入入口。
下面的 Python 示例演示核心控制逻辑。它省略了生产级鉴权、持久化队列和通知发送，但包含环境变量密钥、超时、脱敏、去重以及结构化输出约束：
import hashlib
import json
import os
import re
from datetime import datetime, timedelta, timezone
import httpx
from fastapi import FastAPI, HTTPException, Request
app = FastAPI()
LOKI_URL = os.environ["LOKI_URL"].rstrip("/")
MODEL_BASE_URL = os.environ["MODEL_BASE_URL"].rstrip("/")
MODEL_API_KEY = os.environ["MODEL_API_KEY"]
MODEL_NAME = os.environ["MODEL_NAME"]
ALLOWED_SERVICES = set(os.environ["ALLOWED_SERVICES"].split(","))
PATTERNS = [
(re.compile(r"(?i)authorization[:=]\\s*bearer\\s+[^\\s,]+"), "authorization=[REDACTED]"),
(re.compile(r"(?i)(api[_-]?key|token|password)[:=]\\s*[^\\s,]+"), r"\\1=[REDACTED]"),
(re.compile(r"\\b1[3-9]\\d{9}\\b"), "[PHONE]"),
]
def redact(text: str) -> str:
for pattern, replacement in PATTERNS:
text = pattern.sub(replacement, text)
return text
def fingerprint(alert: dict) -> str:
labels = alert.get("labels", {
})
raw = "|".join([
alert.get("status", ""),
labels.get("alertname", ""),
labels.get("service", ""),
labels.get("env", ""),
alert.get("startsAt", ""),
])
return hashlib.sha256(raw.encode()).hexdigest()
async def query_logs(service: str) -> list[str]:
if service not in ALLOWED_SERVICES:
raise HTTPException(400, "service is not allowed")
end = datetime.now(timezone.utc)
start = end - timedelta(minutes=10)
query = f'{
{job="application",service="{service}",level="error"}}'
params = {
"query": query,
"start": str(int(start.timestamp() * 1_000_000_000)),
"end": str(int(end.timestamp() * 1_000_000_000)),
"limit": "200",
"direction": "backward",
}
async with httpx.AsyncClient(timeout=8.0) as client:
response = await client.get(f"{LOKI_URL}/loki/api/v1/query_range", params=params)
response.raise_for_status()
lines = []
for stream in response.json().get("data", {
}).get("result", []):
for _, line in stream.get("values", []):
cleaned = redact(line)[:2000]
if cleaned not in lines:
lines.append(cleaned)
return lines[:80]
async def summarize(service: str, lines: list[str]) -> dict:
prompt = {
"task": "根据日志生成故障摘要，不得把推测写成已确认事实",
"service": service,
"required_fields": ["observations", "hypotheses", "next_steps", "evidence"],
"logs": lines,
}
payload = {
"model": MODEL_NAME,
"messages": [
{
"role": "system", "content": "只输出 JSON。引用 evidence 时保留对应日志序号。"},
{
"role": "user", "content": json.dumps(prompt, ensure_ascii=False)},
],
"temperature": 0.1,
}
headers = {
"Authorization": f"Bearer {MODEL_API_KEY}"}
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{MODEL_BASE_URL}/chat/completions",
headers=headers,
json=payload,
)
response.raise_for_status()
content = response.json()["choices"][0]["message"]["content"]
return json.loads(content)
@app.post("/alerts")
async def receive_alerts(request: Request):
body = await request.json()
results = []
for alert in body.get("alerts", []):
if alert.get("status") != "firing":
continue
service = alert.get("labels", {
}).get("service", "")
lines = await query_logs(service)
if not lines:
continue
results.append({
"idempotency_key": fingerprint(alert),
"service": service,
"summary": await summarize(service, lines),
})
return {
"results": results}
示例采用常见的 /chat/completions 请求形态，但并不意味着所有服务都支持相同路径、字段或响应结构。接入前必须以所选接口的当前文档为准；如果使用 HaerAPI 作为模型接入端点，也应把地址、模型名和密钥全部放入部署环境变量，并先在非敏感测试日志上验证协议兼容性、超时行为和错误格式。
对应的容器环境变量可以这样配置：
services:
log-summarizer:
image: your-registry/log-summarizer:latest
environment:
LOKI_URL: http://loki:3100
MODEL_BASE_URL: ${
MODEL_BASE_URL}
MODEL_API_KEY: ${
MODEL_API_KEY}
MODEL_NAME: ${
MODEL_NAME}
ALLOWED_SERVICES: order-api,payment-worker
不要将真实密钥写入 Compose 文件或镜像。生产环境应使用现有的 Secret 管理机制注入，并限制摘要服务读取其他业务 Secret 的权限。
第四步：让摘要可以审计和降级
模型返回 JSON 不等于结果可靠。服务端至少应做四项校验：字段是否齐全；内容长度是否超限；evidence 是否只引用已发送的日志序号；输出是否再次出现疑似密钥或个人信息。解析失败时，应保存错误类型并退化为确定性模板，例如“服务、时间窗、匹配数量、Loki 查询链接”，而不是阻断原始告警。
幂等键也不能只计算后返回。生产实现应将它写入带唯一约束的数据库或缓存，并记录处理状态。收到相同键时返回已有结果，避免 Alertmanager 重试导致重复调用。若需要重试模型接口，应只对连接失败、超时或明确可重试的服务端错误进行有限次数退避；参数错误和鉴权失败应立即进入人工可见的失败队列。
建议同时记录以下审计字段：告警指纹、查询时间窗、日志条数、脱敏规则版本、提示词模板版本、模型标识、请求耗时、结果状态和人工修订结果。审计记录中不要再次保存未经脱敏的完整提示词。
常见问题
为什么不让模型直接查询 Loki？
直接授予任意 LogQL 查询能力会扩大数据访问面，也难以控制时间窗和租户边界。更稳妥的方式是由服务端生成白名单查询，只把有限结果交给模型。确有 Agent 查询需求时，也应提供参数化工具，而不是暴露通用查询接口。
正则脱敏足够吗？
通常不够。正则适合处理格式明确的令牌、手机号和邮箱，但无法可靠识别所有自然语言敏感信息。应组合源头禁记、字段级删除、正则替换、访问控制和抽样审计，并根据业务数据类型更新规则。
模型摘要能否直接写“根因”？
除非已有确定证据，否则不应这样设计。输出应区分 observations 与 hypotheses：前者只能陈述日志中可验证的现象，后者必须标明待验证，并给出对应证据和下一步检查方法。
如何控制调用量？
先利用 Alertmanager 分组和抑制，再按告警指纹去重；回查时限制时间窗、日志条数和单行长度。还可以设置每个服务的并发上限与每日预算。达到限制时应发送普通告警，而不是静默丢弃。
为什么告警恢复后不再调用模型？
恢复通知通常只需要附带持续时间和相关事件，不一定需要再次生成摘要。示例跳过了 resolved 状态。若业务需要生成复盘时间线，应把触发和恢复事件写入同一事件记录，再异步汇总，避免在 Webhook 请求内执行长任务。
总结
日志告警接入大模型的关键不在于提示词写得多复杂，而在于职责边界是否明确。Loki 和 LogQL负责筛选可验证事实，Alertmanager 负责聚合与路由，摘要服务负责白名单查询、脱敏、裁剪、幂等和审计，模型只负责在受限材料上生成候选摘要。
落地时应先用非敏感日志验证完整链路，再逐步加入真实服务；先保证模型不可用时原始告警仍能送达，再优化摘要质量。只有当每项结论都能回到日志证据、每次调用都可追踪、敏感数据都有明确处理规则时，这条链路才适合作为值班工作的辅助系统。
