LLM Zoomcamp 端到端项目实战：用 PostgreSQL、Grafana 与 Docker Compose 构建 RAG 应用的可观测性体系

LLM Zoomcamp 端到端项目实战：用 PostgreSQL、Grafana 与 Docker Compose 构建 RAG 应用的可观测性体系

【免费下载链接】llm-zoomcampLLM Zoomcamp - a free online course about real-life applications of LLMs. In 10 weeks you will learn how to build an AI system that answers questions about your knowledge base. Register here 👇🏼项目地址: https://gitcode.com/GitHub_Trending/ll/llm-zoomcamp

导读

本文是 LLM Zoomcamp 模块 07「端到端项目示例」中监控与容器化环节的完整技术指南。我们将把模块 05 的监控方案（LLM-as-a-Judge 评估、PostgreSQL 存储、Grafana 看板）迁移到健身助手（fitness assistant）RAG 应用上，并用 Docker Compose 一次性编排数据库、应用与可视化三件套。读完本文，你将掌握：如何把相关性评估嵌入 RAG 主流程、如何用两张数据库表沉淀会话与用户反馈、如何用三个容器完成从提问到监控的闭环，以及如何通过 Grafana API 自动化配置看板。

监控方案的整体思路

在模块 07 前几课中，我们完成了健身助手的检索评估（评估检索）、RAG 评估（评估 RAG）以及 Flask API 与摄取管线（接口与摄取）。此时应用已经能回答用户关于动作、肌肉群、器械的问题，但我们对线上运行状况一无所知：

• 每条回答是否真的与问题相关？
• 每次调用花费了多少 token、多少钱？
• 用户对回答是否满意？

本课的做法与模块 05 一脉相承，但做了两处关键改造：

1. 把评估从离线搬进线上：模块 05 先离线批量评估再入库（见 03-evaluating-rag.md），这里则把 LLM-as-a-Judge 直接嵌入rag函数，让每一次回答都附带相关性评分；
2. 把组件容器化：用 Docker Compose 把 PostgreSQL、Flask 应用、Grafana 三个服务编排起来，一条docker compose up -d即可启动整条可观测链路。

这条数据流可以概括为：Flask API → RAG（含评估）→ PostgreSQL（会话 + 反馈）→ Grafana（实时看板），全部封装在 Docker Compose 网络中。需要说明的是，模块 07 录制于 2024 年且未再更新，属于可选但依然有价值的实战参考（见模块 README）。

第一步：把 LLM-as-a-Judge 评估嵌入 RAG 流程

模块 05 的评判器把评估封装成了独立的evaluate_relevance函数，并定义了RelevanceVerdict结构化输出模型（源码见 judge.py）。本课用更轻量的方式实现同一目标：通过提示词要求模型输出可解析的 JSON。

评估提示词模板

import json from time import time from openai import OpenAI import ingest openai_client = OpenAI() index = ingest.load_index() evaluation_prompt_template = """ You are an expert evaluator for a RAG system. Your task is to analyze the relevance of the generated answer to the given question. Based on the relevance of the generated answer, you will classify it as 'NON_RELEVANT', 'PARTLY_RELEVANT', or 'RELEVANT'. Here is the data for evaluation: Question: {question} Generated Answer: {answer} Please analyze the content and context of the generated answer in relation to the question and provide your evaluation in parsable JSON without using code blocks: {{ 'Relevance': 'NON_RELEVANT' | 'PARTLY_RELEVANT' | 'RELEVANT', 'Explanation': '[Provide a brief explanation for your evaluation]' }} """.strip()

提示词要求模型输出三档结论并附简短解释：

与模块 05 的judge_instructions（judge.py）相比，这里把指令和问题、答案合并在同一个模板里，并用"不使用代码块、输出可解析 JSON"的约束来换取简单直接的解析逻辑。

评估函数与成本计算

def evaluate_relevance(question, answer): prompt = evaluation_prompt_template.format(question=question, answer=answer) evaluation, tokens = llm(prompt, model="gpt-5.4-mini") try: json_eval = json.loads(evaluation) return json_eval, tokens except json.JSONDecodeError: result = {"Relevance": "UNKNOWN", "Explanation": "Failed to parse evaluation"} return result, tokens def calculate_openai_cost(model, tokens): openai_cost = 0 if "gpt-5.4-mini" in model: openai_cost = ( tokens["prompt_tokens"] * 0.15 + tokens["completion_tokens"] * 0.60 ) / 1_000_000 return openai_cost

两个细节值得注意：

• 容错设计：json.loads失败时不会让整个请求崩溃，而是返回UNKNOWN档位和"解析失败"的说明，保证监控链路不断；
• 成本模型：gpt-5.4-mini按输入每百万 token 0.15 美元、输出每百万 token 0.60 美元计费。这与模块 05 的 metrics.py 中calculate_cost的计费口径完全一致，说明这套成本估算在课程内是统一标准。

更新后的 rag 函数

def rag(query, model="gpt-5.4-mini"): t0 = time() search_results = search(query) prompt = build_prompt(query, search_results) answer, token_stats = llm(prompt, model=model) relevance, rel_token_stats = evaluate_relevance(query, answer) t1 = time() took = t1 - t0 openai_cost_rag = calculate_openai_cost(model, token_stats) openai_cost_eval = calculate_openai_cost(model, rel_token_stats) openai_cost = openai_cost_rag + openai_cost_eval return { "answer": answer, "model_used": model, "response_time": took, "relevance": relevance.get("Relevance", "UNKNOWN"), "relevance_explanation": relevance.get("Explanation", "Failed to parse"), "prompt_tokens": token_stats["prompt_tokens"], "completion_tokens": token_stats["completion_tokens"], "total_tokens": token_stats["total_tokens"], "eval_prompt_tokens": rel_token_stats["prompt_tokens"], "eval_completion_tokens": rel_token_stats["completion_tokens"], "eval_total_tokens": rel_token_stats["total_tokens"], "openai_cost": openai_cost, }

升级后的rag返回结构比模块 04 的基础版（见 04-interface.md）多出一整套监控字段：相关性档位、相关性解释、评估调用的 token 消耗，以及 RAG 调用与评估调用合并后的总成本。返回的 dict 结构与conversations表一一对应，为下一步入库铺好了路。

第二步：数据库函数——用 PostgreSQL 沉淀会话与反馈

模块 05 用 db_init.py 定义了conversations与feedback两张表，本课沿用同一设计，并针对健身助手项目做了字段调整。

连接配置与时区

import os import psycopg2 from psycopg2.extras import DictCursor from datetime import datetime from zoneinfo import ZoneInfo TZ_INFO = os.getenv("TZ", "Europe/Berlin") tz = ZoneInfo(TZ_INFO) def get_db_connection(): return psycopg2.connect( host=os.getenv("POSTGRES_HOST", "postgres"), database=os.getenv("POSTGRES_DB", "fitness_assistant"), user=os.getenv("POSTGRES_USER", "user"), password=os.getenv("POSTGRES_PASSWORD", "password"), )

连接参数全部通过环境变量注入，并带有合理的默认值。注意默认POSTGRES_HOST=postgres——这正是 Docker Compose 中服务名，本地开发时则要覆盖为localhost。时区默认Europe/Berlin，可用TZ环境变量覆盖，所有时间戳都带时区信息（TIMESTAMP WITH TIME ZONE），这与模块 05 强调的"时区感知时间戳"实践一致。

建表：conversations 与 feedback

def init_db(): conn = get_db_connection() try: with conn.cursor() as cur: cur.execute("DROP TABLE IF EXISTS feedback") cur.execute("DROP TABLE IF EXISTS conversations") cur.execute(""" CREATE TABLE conversations ( id TEXT PRIMARY KEY, question TEXT NOT NULL, answer TEXT NOT NULL, model_used TEXT NOT NULL, response_time FLOAT NOT NULL, relevance TEXT NOT NULL, relevance_explanation TEXT NOT NULL, prompt_tokens INTEGER NOT NULL, completion_tokens INTEGER NOT NULL, total_tokens INTEGER NOT NULL, eval_prompt_tokens INTEGER NOT NULL, eval_completion_tokens INTEGER NOT NULL, eval_total_tokens INTEGER NOT NULL, openai_cost FLOAT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL ) """) cur.execute(""" CREATE TABLE feedback ( id SERIAL PRIMARY KEY, conversation_id TEXT REFERENCES conversations(id), feedback INTEGER NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL ) """) conn.commit() finally: conn.close()

两张表的职责划分非常清晰：

• conversations：一条记录就是一次完整问答，主键id为应用侧生成的 UUID（TEXT 类型），并冗余存储全部监控指标——相关性、两段调用的 token、总成本、响应时间；
• feedback：通过外键conversation_id关联到具体会话，feedback字段记录用户打分。这里只允许1或-1（赞/踩），比模块 05 中同时容纳 judge 与 user 两类来源（见 db_init.py）更聚焦于用户反馈这一种场景。

init_db先DROP再CREATE，适合开发期反复初始化；finally保证连接无论如何都会关闭。

写入函数：save_conversation 与 save_feedback

def save_conversation(conversation_id, question, answer_data, timestamp=None): if timestamp is None: timestamp = datetime.now(tz) conn = get_db_connection() try: with conn.cursor() as cur: cur.execute( """ INSERT INTO conversations (id, question, answer, model_used, response_time, relevance, relevance_explanation, prompt_tokens, completion_tokens, total_tokens, eval_prompt_tokens, eval_completion_tokens, eval_total_tokens, openai_cost, timestamp) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( conversation_id, question, answer_data["answer"], answer_data["model_used"], answer_data["response_time"], answer_data["relevance"], answer_data["relevance_explanation"], answer_data["prompt_tokens"], answer_data["completion_tokens"], answer_data["total_tokens"], answer_data["eval_prompt_tokens"], answer_data["eval_completion_tokens"], answer_data["eval_total_tokens"], answer_data["openai_cost"], timestamp ), ) conn.commit() finally: conn.close()

save_conversation直接接收rag返回的 dict（answer_data），通过键名取值完成 15 个字段的批量写入，使用%s占位符参数化查询防止 SQL 注入。

def save_feedback(conversation_id, feedback, timestamp=None): if timestamp is None: timestamp = datetime.now(tz) conn = get_db_connection() try: with conn.cursor() as cur: cur.execute( "INSERT INTO feedback (conversation_id, feedback, timestamp) VALUES (%s, %s, %s)", (conversation_id, feedback, timestamp), ) conn.commit() finally: conn.close()

save_feedback则只需三个字段，按会话主键记录一条点赞或点踩。两个函数都允许外部传入timestamp，便于测试或数据回填。

第三步：更新 Flask API——记录问答、接收反馈

有了rag与db两个模块，Flask API 只需做两件事：每次回答都落库，并提供反馈入口。

/question 端点：每次问答自动存档

import uuid from flask import Flask, request, jsonify from rag import rag import db app = Flask(__name__) @app.route("/question", methods=["POST"]) def handle_question(): data = request.json question = data["question"] if not question: return jsonify({"error": "No question provided"}), 400 conversation_id = str(uuid.uuid4()) answer_data = rag(question) db.save_conversation( conversation_id=conversation_id, question=question, answer_data=answer_data, ) result = { "conversation_id": conversation_id, "question": question, "answer": answer_data["answer"], } return jsonify(result)

相比模块 04 的原始版本（04-interface.md），核心变化是：生成conversation_id后不再只返回答案，而是把完整的answer_data交给db.save_conversation持久化。对外返回时仍然只暴露conversation_id、question、answer三个字段，监控细节留在数据库内部。

/feedback 端点：接收用户赞踩

@app.route("/feedback", methods=["POST"]) def handle_feedback(): data = request.json conversation_id = data["conversation_id"] feedback = data["feedback"] if not conversation_id or feedback not in [1, -1]: return jsonify({"error": "Invalid input"}), 400 db.save_feedback( conversation_id=conversation_id, feedback=feedback, ) return jsonify({"message": f"Feedback received: {feedback}"}) if __name__ == "__main__": app.run(debug=True, host="0.0.0.0", port=5000)

/feedback端点在服务端校验两件事：conversation_id非空、feedback必须是1或-1，非法输入直接返回 400。这保证了入库数据的质量，也避免了客户端把任意整数塞进数据库。

安装数据库驱动

Flask API 依赖 PostgreSQL 驱动，用 uv 添加：

uv add psycopg2-binary

psycopg2-binary是预编译版本，免去本地编译依赖，适合在 Docker 等干净环境里直接安装。

第四步：Docker Compose 容器化

这是本课的核心产出：把 PostgreSQL、应用、Grafana 三个服务编排进一个docker-compose.yaml，实现"一条命令启动整条监控链路"。

编排文件

services: postgres: image: postgres:16 environment: POSTGRES_DB: fitness_assistant POSTGRES_USER: user POSTGRES_PASSWORD: password ports: - "5432:5432" volumes: - postgres_data:/var/lib/postgresql/data app: build: context: . dockerfile: Dockerfile environment: - POSTGRES_HOST=postgres - POSTGRES_DB=fitness_assistant - POSTGRES_USER=user - POSTGRES_PASSWORD=password - OPENAI_API_KEY=${OPENAI_API_KEY} ports: - "5000:5000" depends_on: - postgres grafana: image: grafana/grafana:latest ports: - "3000:3000" environment: - GF_SECURITY_ADMIN_PASSWORD=admin depends_on: - postgres volumes: postgres_data:

三个服务各司其职：

几个要点：

• 服务发现：app与grafana通过服务名postgres访问数据库，这正是前面get_db_connection默认POSTGRES_HOST=postgres的原因；
• 依赖顺序：depends_on: postgres保证数据库先启动，但注意 Compose 只保证启动顺序，不保证数据库已就绪，app内部仍需有重试或由运维手动初始化；
• 密钥管理：OPENAI_API_KEY=${OPENAI_API_KEY}从宿主机环境读取，配合.env文件使用，密钥不写死在镜像里；
• 数据持久化：postgres_data命名卷挂载到/var/lib/postgresql/data，容器重建后数据不丢。

对比模块 05 用 Makefile +docker run手工管理容器的方式（见 Makefile，那里用的是postgres:17），Docker Compose 把所有参数声明式地收拢到一个文件里，可复现性更好。

Dockerfile：基于 uv 的多阶段构建

FROM python:3.14-slim COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/ WORKDIR /app ENV PATH="/app/.venv/bin:$PATH" COPY pyproject.toml uv.lock .python-version ./ RUN uv sync --locked COPY . . EXPOSE 5000 CMD ["python", "app.py"]

这个 Dockerfile 充分利用了 uv 的优势：

• 从ghcr.io/astral-sh/uv镜像直接拷贝 uv 二进制，无需在镜像里安装 Rust 工具链；
• 先只拷贝pyproject.toml、uv.lock、.python-version三个文件执行uv sync --locked，充分利用 Docker 层缓存——依赖没变时这层缓存不会被击穿；
• --locked严格按照uv.lock锁定版本安装，保证构建可复现；
• 虚拟环境路径/app/.venv/bin加入PATH，后续python app.py直接使用项目环境。

环境变量文件与启动流程

.env文件只需提供 API 密钥：

OPENAI_API_KEY=your-key-here

完整启动流程分三步：

docker compose up -d

后台启动三个容器。然后初始化数据库（建两张表）：

uv run python -c "import db; db.init_db()"

最后访问服务：

• 应用：http://localhost:5000（可先curl -X POST http://localhost:5000/question -H "Content-Type: application/json" -d '{"question": "What exercises target the chest?"}'验证）
• Grafana：http://localhost:3000（默认管理员账号admin/admin）

第五步：Grafana 看板——从数据到洞察

数据落库只是第一步，真正让监控产生价值的是看板。模块 05 用 Grafana 直连 PostgreSQL 构建了完整的监控面板（详见 12-grafana.md），本课直接复用这套查询思路，只是把表名与字段对齐到健身助手的conversations与feedback表。

核心查询模式

看板的每个面板本质都是一条针对 PostgreSQL 的 SQL。模块 05 总结了两个关键习惯，同样适用于本项目：

1. 时间列别名time：Grafana 靠AS time识别 x 轴时间列；
2. 使用$__timeFrom()/$__timeTo()/$__timeGroup()变量：让面板跟随顶部选择的时间范围，并对时间序列按区间聚合。

典型面板对应的 SQL 如下：

响应时间面板（每次问答的 LLM 耗时，直接绘制原始值）：

SELECT timestamp AS time, response_time FROM conversations WHERE timestamp BETWEEN $__timeFrom() AND $__timeTo() ORDER BY timestamp

Token 用量面板（按时间桶求平均，避免长时间范围点数过多）：

SELECT $__timeGroup(timestamp, $__interval) AS time, AVG(total_tokens) AS avg_tokens FROM conversations WHERE timestamp BETWEEN $__timeFrom() AND $__timeTo() GROUP BY 1 ORDER BY 1

成本面板（按时间桶求和）：

SELECT $__timeGroup(timestamp, $__interval) AS time, SUM(cost) AS total_cost FROM conversations WHERE timestamp BETWEEN $__timeFrom() AND $__timeTo() AND cost > 0 GROUP BY 1 ORDER BY 1

相关性分布面板（统计 RELEVANT / PARTLY_RELEVANT / NON_RELEVANT 各占多少，用饼图）：

SELECT relevance, COUNT(*) as count FROM conversations WHERE timestamp BETWEEN $__timeFrom() AND $__timeTo() GROUP BY relevance

用户反馈面板（赞 vs 踩）：

SELECT SUM(CASE WHEN feedback > 0 THEN 1 ELSE 0 END) as thumbs_up, SUM(CASE WHEN feedback < 0 THEN 1 ELSE 0 END) as thumbs_down FROM feedback WHERE timestamp BETWEEN $__timeFrom() AND $__timeTo()

模块 05 还给出了推荐布局：顶行放最近会话表格（宽），中间行放模型使用柱状图 + 相关性饼图，底行放响应时间、Token 用量、成本三个时序图。本项目的面板集完全可以在同一布局上落地。

用 Grafana API 自动化配置

手工在界面上点选数据源和面板既繁琐又不可复现。最终项目提供了一个自动化脚本（位于grafana/目录的init.py），核心流程是：

1. 调用 Grafana API 创建 PostgreSQL 数据源（Host 指向postgres:5432，数据库fitness_assistant，用户/密码与 Compose 一致）；
2. 加载预先写好的 dashboard JSON 文件，通过 API 导入看板。

这种"配置即代码"的方式让整个监控体系可以随项目一起版本化、可复现，也是模块 07 最终成品比视频演示更进一步的地方（模块 README 明确提到最终项目"better README、code readability、automated Grafana provisioning"，见模块 07 README）。

小结：一条完整的可观测闭环

回顾整条链路，本课完成了从"能回答问题"到"可观测、可评价、可迭代"的跨越：

1. 评估内嵌：rag每次调用都产出相关性档位、解释与成本；
2. 数据沉淀：conversations记录每次问答的完整指标，feedback记录用户赞踩，两者通过 UUID 关联；
3. API 对接：/question自动存档，/feedback接收用户反馈；
4. 一键部署：Docker Compose 编排 PostgreSQL、应用、Grafana，docker compose up -d即可起服务；
5. 看板呈现：Grafana 直连 PostgreSQL，用 SQL 面板实时展示响应时间、Token、成本、相关性与用户反馈，并可用 API 脚本自动化配置。

这套模式与模块 05 的课程助教项目（见模块 05 目录）形成对照：表结构、成本公式、看板查询都同源同构，区别仅在于领域数据不同。如果你正在准备自己的课程项目，完全可以把它当作可复用的监控脚手架——把conversations表的字段映射到你自己的答案结构，把看板 SQL 里的表名和字段名替换为你的项目表，监控体系即可快速迁移。

【免费下载链接】llm-zoomcampLLM Zoomcamp - a free online course about real-life applications of LLMs. In 10 weeks you will learn how to build an AI system that answers questions about your knowledge base. Register here 👇🏼项目地址: https://gitcode.com/GitHub_Trending/ll/llm-zoomcamp