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 assistantRAG 应用上并用 Docker Compose 一次性编排数据库、应用与可视化三件套。读完本文你将掌握如何把相关性评估嵌入 RAG 主流程、如何用两张数据库表沉淀会话与用户反馈、如何用三个容器完成从提问到监控的闭环以及如何通过 Grafana API 自动化配置看板。监控方案的整体思路在模块 07 前几课中我们完成了健身助手的检索评估评估检索、RAG 评估评估 RAG以及 Flask API 与摄取管线接口与摄取。此时应用已经能回答用户关于动作、肌肉群、器械的问题但我们对线上运行状况一无所知每条回答是否真的与问题相关每次调用花费了多少 token、多少钱用户对回答是否满意本课的做法与模块 05 一脉相承但做了两处关键改造把评估从离线搬进线上模块 05 先离线批量评估再入库见 03-evaluating-rag.md这里则把 LLM-as-a-Judge 直接嵌入rag函数让每一次回答都附带相关性评分把组件容器化用 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()提示词要求模型输出三档结论并附简短解释档位含义RELEVANT回答直接回应了问题PARTLY_RELEVANT回答部分回应了问题NON_RELEVANT回答与问题无关与模块 05 的judge_instructionsjudge.py相比这里把指令和问题、答案合并在同一个模板里并用不使用代码块、输出可解析 JSON的约束来换取简单直接的解析逻辑。评估函数与成本计算def evaluate_relevance(question, answer): prompt evaluation_prompt_template.format(questionquestion, answeranswer) evaluation, tokens llm(prompt, modelgpt-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, modelgpt-5.4-mini): t0 time() search_results search(query) prompt build_prompt(query, search_results) answer, token_stats llm(prompt, modelmodel) 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( hostos.getenv(POSTGRES_HOST, postgres), databaseos.getenv(POSTGRES_DB, fitness_assistant), useros.getenv(POSTGRES_USER, user), passwordos.getenv(POSTGRES_PASSWORD, password), )连接参数全部通过环境变量注入并带有合理的默认值。注意默认POSTGRES_HOSTpostgres——这正是 Docker Compose 中服务名本地开发时则要覆盖为localhost。时区默认Europe/Berlin可用TZ环境变量覆盖所有时间戳都带时区信息TIMESTAMP WITH TIME ZONE这与模块 05 强调的时区感知时间戳实践一致。建表conversations 与 feedbackdef 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为应用侧生成的 UUIDTEXT 类型并冗余存储全部监控指标——相关性、两段调用的 token、总成本、响应时间feedback通过外键conversation_id关联到具体会话feedback字段记录用户打分。这里只允许1或-1赞/踩比模块 05 中同时容纳 judge 与 user 两类来源见 db_init.py更聚焦于用户反馈这一种场景。init_db先DROP再CREATE适合开发期反复初始化finally保证连接无论如何都会关闭。写入函数save_conversation 与 save_feedbackdef save_conversation(conversation_id, question, answer_data, timestampNone): 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返回的 dictanswer_data通过键名取值完成 15 个字段的批量写入使用%s占位符参数化查询防止 SQL 注入。def save_feedback(conversation_id, feedback, timestampNone): 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_idconversation_id, questionquestion, answer_dataanswer_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_idconversation_id, feedbackfeedback, ) return jsonify({message: fFeedback received: {feedback}}) if __name__ __main__: app.run(debugTrue, host0.0.0.0, port5000)/feedback端点在服务端校验两件事conversation_id非空、feedback必须是1或-1非法输入直接返回 400。这保证了入库数据的质量也避免了客户端把任意整数塞进数据库。安装数据库驱动Flask API 依赖 PostgreSQL 驱动用 uv 添加uv add psycopg2-binarypsycopg2-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_HOSTpostgres - POSTGRES_DBfitness_assistant - POSTGRES_USERuser - POSTGRES_PASSWORDpassword - OPENAI_API_KEY${OPENAI_API_KEY} ports: - 5000:5000 depends_on: - postgres grafana: image: grafana/grafana:latest ports: - 3000:3000 environment: - GF_SECURITY_ADMIN_PASSWORDadmin depends_on: - postgres volumes: postgres_data:三个服务各司其职服务镜像/构建对外端口关键配置postgrespostgres:165432数据库名/用户/密码、命名卷postgres_data持久化数据app本地 Dockerfile 构建5000通过POSTGRES_HOSTpostgres指向数据库服务OPENAI_API_KEY从宿主机.env注入grafanagrafana/grafana:latest3000管理员密码设为admin几个要点服务发现app与grafana通过服务名postgres访问数据库这正是前面get_db_connection默认POSTGRES_HOSTpostgres的原因依赖顺序depends_on: postgres保证数据库先启动但注意 Compose 只保证启动顺序不保证数据库已就绪app内部仍需有重试或由运维手动初始化密钥管理OPENAI_API_KEY${OPENAI_API_KEY}从宿主机环境读取配合.env文件使用密钥不写死在镜像里数据持久化postgres_data命名卷挂载到/var/lib/postgresql/data容器重建后数据不丢。对比模块 05 用 Makefile docker run手工管理容器的方式见 Makefile那里用的是postgres:17Docker Compose 把所有参数声明式地收拢到一个文件里可复现性更好。Dockerfile基于 uv 的多阶段构建FROM python:3.14-slim COPY --fromghcr.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_KEYyour-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?}验证Grafanahttp://localhost:3000默认管理员账号admin/admin第五步Grafana 看板——从数据到洞察数据落库只是第一步真正让监控产生价值的是看板。模块 05 用 Grafana 直连 PostgreSQL 构建了完整的监控面板详见 12-grafana.md本课直接复用这套查询思路只是把表名与字段对齐到健身助手的conversations与feedback表。核心查询模式看板的每个面板本质都是一条针对 PostgreSQL 的 SQL。模块 05 总结了两个关键习惯同样适用于本项目时间列别名timeGrafana 靠AS time识别 x 轴时间列使用$__timeFrom()/$__timeTo()/$__timeGroup()变量让面板跟随顶部选择的时间范围并对时间序列按区间聚合。典型面板对应的 SQL 如下响应时间面板每次问答的 LLM 耗时直接绘制原始值SELECT timestamp AS time, response_time FROM conversations WHERE timestamp BETWEEN $__timeFrom() AND $__timeTo() ORDER BY timestampToken 用量面板按时间桶求平均避免长时间范围点数过多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核心流程是调用 Grafana API 创建 PostgreSQL 数据源Host 指向postgres:5432数据库fitness_assistant用户/密码与 Compose 一致加载预先写好的 dashboard JSON 文件通过 API 导入看板。这种配置即代码的方式让整个监控体系可以随项目一起版本化、可复现也是模块 07 最终成品比视频演示更进一步的地方模块 README 明确提到最终项目better README、code readability、automated Grafana provisioning见模块 07 README。小结一条完整的可观测闭环回顾整条链路本课完成了从能回答问题到可观测、可评价、可迭代的跨越评估内嵌rag每次调用都产出相关性档位、解释与成本数据沉淀conversations记录每次问答的完整指标feedback记录用户赞踩两者通过 UUID 关联API 对接/question自动存档/feedback接收用户反馈一键部署Docker Compose 编排 PostgreSQL、应用、Grafanadocker compose up -d即可起服务看板呈现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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考