从零构建工程师式AI工作流：以博客评估Agent为例

在实际 AI 应用开发中我们常常面临一个困境一次性的、零散的提示词Prompt难以构建稳定、可复用、可维护的智能任务。无论是构建一个复杂的 AI 客服还是一个需要多步骤推理的数据分析 Agent临时编写的提示词往往脆弱、难以调试且无法沉淀为团队资产。这正是“工程师式 AI 工作流”概念试图解决的问题。它倡导将 AI 能力像传统软件工程一样通过模块化、流程化、版本化的“蓝图”来构建而非依赖临时的、一次性的魔法咒语。本文将以一个名为AI Blueprint的开源项目为引子探讨如何将工程师思维融入 AI 应用开发。我们将从概念入手理解工作流与一次性提示的本质区别然后通过一个具体的、可运行的示例项目展示如何设计、实现并部署一个工程师式的 AI 工作流。无论你是希望提升现有 AI 应用稳定性的开发者还是对构建复杂 AI Agent 感兴趣的工程师本文都将提供一个从理论到实践的完整路径。1. 理解工程师式 AI 工作流从“咒语”到“蓝图”在深入代码之前我们必须先厘清核心概念。一次性提示词就像给 AI 下达的一条指令例如“总结这篇文章”。它简单直接但缺乏结构、容错性和复用性。当任务变复杂比如“阅读这篇技术文章提取其中的代码片段评估其安全性并生成一份带改进建议的报告”时单一提示词就会力不从心导致输出不稳定、遗漏步骤或逻辑混乱。工程师式 AI 工作流则不同它将复杂任务拆解为一系列定义良好的、可连接的步骤或称为“节点”。每个步骤都有明确的输入、处理逻辑可能包含一个或多个子提示词、函数调用和输出。步骤之间通过数据流连接形成一个有向无环图DAG。这种模式带来了几个关键优势模块化与复用每个处理步骤如“文本提取”、“代码分析”、“报告生成”都可以独立开发、测试并在不同工作流中复用。可观测性与调试每个步骤的输入、输出、执行状态和耗时都可以被记录和追踪使得调试 AI 应用像调试普通程序一样清晰。稳定性与鲁棒性可以针对每个步骤设计错误处理、重试逻辑和降级方案避免因单个环节失败导致整个流程崩溃。团队协作与版本控制工作流定义蓝图可以作为代码文件如 YAML、JSON进行版本管理方便团队协作和迭代。当前许多平台和框架都在向这个方向演进例如 LangChain、LlamaIndex 提供了构建链Chain和代理Agent的基础能力Dify、Coze 等平台提供了可视化的低代码工作流编排界面。而AI Blueprint这类项目则更侧重于提供一种工程化的实践范式和可参考的实现模板强调代码结构、配置管理和部署运维。2. 环境准备与项目初始化为了演示一个完整的工程师式 AI 工作流我们将构建一个简单的“技术博客质量评估 Agent”。这个工作流会接收一篇博客文章的 URL经过“内容抓取”、“关键信息提取”、“多维度评估”和“报告生成”四个步骤最终输出一份结构化的评估报告。2.1 技术栈与工具选择我们将选择一个轻量级但足够表达工作流思想的组合Python 3.9: 作为主要开发语言。LangChain: 用于构建链式调用和与 LLM 交互。它提供了丰富的工具和抽象是构建 AI 工作流的利器。FastAPI: 用于将工作流封装成 HTTP API 服务便于集成和调用。Pydantic: 用于定义严格的数据模型确保工作流中各个步骤间数据传递的类型安全。Playwright或BeautifulSoup4: 用于网页内容抓取。OpenAI API或本地大模型通过 Ollama、vLLM 等: 作为 LLM 能力提供者。为了演示的通用性我们将使用 OpenAI API 格式的接口。注意选择 LangChain 是因为其生态成熟能清晰展示模块化思想。你也可以使用更轻量的直接 HTTP 调用或探索其他框架如 Semantic Kernel。2.2 创建项目结构与核心依赖首先创建一个标准的 Python 项目目录。mkdir ai_blueprint_demo cd ai_blueprint_demo python -m venv venv # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate创建requirements.txt文件定义项目依赖# 核心框架与API fastapi0.104.1 uvicorn[standard]0.24.0 pydantic2.5.0 langchain0.0.340 langchain-openai0.0.2 langchain-community0.0.10 # 网页抓取与数据处理 playwright1.40.0 beautifulsoup44.12.2 markdownify0.11.6 # 工具类 python-dotenv1.0.0 loguru0.7.2安装依赖并初始化 Playwright 浏览器pip install -r requirements.txt playwright install chromium项目目录结构规划如下这体现了关注点分离的工程思想ai_blueprint_demo/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口 │ ├── workflows/ # 工作流定义目录 │ │ ├── __init__.py │ │ └── blog_evaluator.py # 博客评估工作流 │ ├── nodes/ # 工作流节点步骤实现 │ │ ├── __init__.py │ │ ├── fetcher.py # 抓取节点 │ │ ├── extractor.py # 提取节点 │ │ ├── evaluator.py # 评估节点 │ │ └── reporter.py # 报告节点 │ ├── schemas/ # Pydantic 数据模型 │ │ ├── __init__.py │ │ └── models.py │ └── config.py # 配置文件 ├── .env.example # 环境变量示例 ├── requirements.txt └── README.md3. 实现博客评估工作流从节点到蓝图现在我们开始实现“技术博客质量评估”工作流的各个模块。我们将采用自底向上的方式先实现每个独立的节点再将它们组装成完整的工作流。3.1 定义数据模型Schemas在app/schemas/models.py中定义工作流中流转的数据结构。这是保证类型安全和接口清晰的关键。from pydantic import BaseModel, HttpUrl from typing import List, Optional, Dict, Any from enum import Enum class EvaluationDimension(str, Enum): READABILITY 可读性 TECHNICAL_DEPTH 技术深度 CODE_QUALITY 代码质量 PRACTICALITY 实践性 STRUCTURE 结构清晰度 class EvaluationResult(BaseModel): dimension: EvaluationDimension score: int # 1-10分 comment: str class BlogContent(BaseModel): url: HttpUrl title: str raw_html: Optional[str] None cleaned_text: Optional[str] None code_snippets: List[str] [] metadata: Dict[str, Any] {} # 如作者、发布时间等 class BlogEvaluationReport(BaseModel): blog: BlogContent overall_score: float dimension_scores: List[EvaluationResult] summary: str suggestions: List[str]3.2 实现工作流节点Nodes每个节点都是一个独立的、功能单一的类或函数。我们创建四个节点。节点1内容抓取器 (app/nodes/fetcher.py)这个节点负责从给定的 URL 抓取 HTML 内容。import logging from playwright.sync_api import sync_playwright from app.schemas.models import BlogContent from typing import Optional logger logging.getLogger(__name__) class ContentFetcher: def __init__(self, timeout_ms: int 30000): self.timeout timeout_ms def run(self, blog_content: BlogContent) - BlogContent: 执行抓取将原始HTML存入blog_content.raw_html url str(blog_content.url) logger.info(fFetching content from {url}) try: with sync_playwright() as p: # 使用无头浏览器更好地处理JS渲染的页面 browser p.chromium.launch(headlessTrue) page browser.new_page() page.goto(url, timeoutself.timeout) # 等待页面主要内容加载可根据实际情况调整选择器 page.wait_for_load_state(networkidle) html_content page.content() browser.close() blog_content.raw_html html_content logger.info(fSuccessfully fetched content, length: {len(html_content)}) except Exception as e: logger.error(fFailed to fetch content from {url}: {e}) # 工作流中应设计错误处理例如重试或使用备用方案 raise return blog_content节点2信息提取器 (app/nodes/extractor.py)这个节点负责从原始 HTML 中提取标题、纯文本和代码片段。import logging from bs4 import BeautifulSoup from markdownify import markdownify as md from app.schemas.models import BlogContent import re logger logging.getLogger(__name__) class InformationExtractor: def run(self, blog_content: BlogContent) - BlogContent: 从raw_html中提取信息填充到blog_content其他字段 if not blog_content.raw_html: raise ValueError(raw_html is empty, cannot extract information.) soup BeautifulSoup(blog_content.raw_html, html.parser) # 提取标题 title_tag soup.find(title) or soup.find(h1) blog_content.title title_tag.get_text().strip() if title_tag else Unknown Title # 提取主要文章内容假设文章在article或main标签内这是一个简化策略 article soup.find(article) or soup.find(main) or soup.find(body) if article: # 将HTML转换为更干净的Markdown文本 cleaned_md md(str(article), heading_styleATX) blog_content.cleaned_text cleaned_md else: blog_content.cleaned_text # 提取代码片段 (假设代码在precode或code标签内) code_blocks soup.find_all([pre, code]) extracted_snippets [] for block in code_blocks: text block.get_text().strip() if text and len(text) 10: # 简单过滤掉太短的片段 extracted_snippets.append(text) blog_content.code_snippets extracted_snippets logger.info(fExtracted: Title{blog_content.title}, Snippets{len(extracted_snippets)}) return blog_content节点3多维评估器 (app/nodes/evaluator.py)这是核心的 AI 环节。我们使用 LangChain 调用 LLM按照预定义的维度进行评估。import logging from langchain.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain.output_parsers import PydanticOutputParser from app.schemas.models import BlogContent, EvaluationResult, EvaluationDimension from typing import List import os from dotenv import load_dotenv load_dotenv() logger logging.getLogger(__name__) class MultiDimensionEvaluator: def __init__(self): # 初始化LLM这里使用OpenAI。可替换为其他兼容API的模型。 self.llm ChatOpenAI( modelgpt-3.5-turbo, temperature0.1, # 低温度保证评估稳定性 api_keyos.getenv(OPENAI_API_KEY) ) # 使用PydanticOutputParser确保LLM输出结构化数据 self.parser PydanticOutputParser(pydantic_objectEvaluationResult) # 构建评估提示词模板 self.evaluation_prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深技术博客评审专家。请根据提供的博客文本和代码片段从以下维度进行客观评估。\n{format_instructions}), (human, 博客标题{title}\n\n博客正文部分{text_preview}\n\n相关代码片段{code_snippets}\n\n请对{dimension}维度进行打分1-10分并给出简短评语。) ]) def _evaluate_single_dimension(self, blog: BlogContent, dimension: EvaluationDimension) - EvaluationResult: 评估单个维度 # 准备输入 text_preview blog.cleaned_text[:1500] if blog.cleaned_text else # 限制长度 code_preview \n---\n.join(blog.code_snippets[:3]) # 取前3个代码片段 prompt self.evaluation_prompt.format_prompt( format_instructionsself.parser.get_format_instructions(), titleblog.title, text_previewtext_preview, code_snippetscode_preview, dimensiondimension.value ) response self.llm.invoke(prompt.to_messages()) try: result self.parser.parse(response.content) result.dimension dimension # 确保维度一致 return result except Exception as e: logger.error(fFailed to parse LLM output for dimension {dimension}: {e}) # 返回一个默认的失败结果在实际项目中应有更完善的降级策略 return EvaluationResult(dimensiondimension, score5, comment评估解析失败) def run(self, blog_content: BlogContent) - List[EvaluationResult]: 对博客进行多维度评估 logger.info(fStarting multi-dimension evaluation for: {blog_content.title}) results [] for dimension in EvaluationDimension: logger.debug(fEvaluating dimension: {dimension}) result self._evaluate_single_dimension(blog_content, dimension) results.append(result) logger.info(Multi-dimension evaluation completed.) return results节点4报告生成器 (app/nodes/reporter.py)这个节点汇总所有评估结果生成一份最终的综合报告。import logging from langchain.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from app.schemas.models import BlogContent, BlogEvaluationReport, EvaluationResult from typing import List import os logger logging.getLogger(__name__) class ReportGenerator: def __init__(self): self.llm ChatOpenAI( modelgpt-3.5-turbo, temperature0.7, # 稍高的温度让总结和建议更具创造性 api_keyos.getenv(OPENAI_API_KEY) ) self.report_prompt ChatPromptTemplate.from_messages([ (system, 你是一位技术主编。请根据以下对一篇技术博客的详细评估结果生成一份综合评估报告。报告需包含一个总体评分基于各维度得分的加权平均满分10分、一段总结性文字和三条具体的改进建议。), (human, 博客标题{title}\n博客链接{url}\n\n各维度评估结果\n{assessment_details}\n\n请生成报告。) ]) def run(self, blog_content: BlogContent, evaluation_results: List[EvaluationResult]) - BlogEvaluationReport: 生成最终评估报告 logger.info(fGenerating final report for: {blog_content.title}) # 计算总体平均分 overall_score sum([r.score for r in evaluation_results]) / len(evaluation_results) # 格式化评估详情 details_str \n.join([f- {r.dimension.value}: {r.score}分。评语{r.comment} for r in evaluation_results]) # 调用LLM生成总结和建议 prompt self.report_prompt.format_prompt( titleblog_content.title, urlstr(blog_content.url), assessment_detailsdetails_str ) response self.llm.invoke(prompt.to_messages()) llm_output response.content # 简单解析LLM输出在实际项目中应使用更稳健的解析如再次使用PydanticOutputParser lines llm_output.split(\n) summary suggestions [] current_section None for line in lines: if 总结 in line or Summary in line: current_section summary elif 建议 in line or Suggestions in line: current_section suggestions elif current_section summary and line.strip() and not line.startswith(-): summary line.strip() elif current_section suggestions and line.strip().startswith(-): suggestions.append(line.strip()[1:].strip()) if not suggestions: suggestions [建议部分解析失败请查看原始评估维度结果。] # 构建最终报告对象 report BlogEvaluationReport( blogblog_content, overall_scoreround(overall_score, 2), dimension_scoresevaluation_results, summarysummary if summary else AI生成总结失败。, suggestionssuggestions[:3] # 最多取三条 ) logger.info(fReport generated. Overall score: {report.overall_score}) return report3.3 组装工作流蓝图 (app/workflows/blog_evaluator.py)现在我们将上述节点按照逻辑顺序组装起来形成一个完整的工作流。这是“蓝图”的核心。import logging from app.schemas.models import BlogContent, BlogEvaluationReport from app.nodes.fetcher import ContentFetcher from app.nodes.extractor import InformationExtractor from app.nodes.evaluator import MultiDimensionEvaluator from app.nodes.reporter import ReportGenerator logger logging.getLogger(__name__) class BlogEvaluationWorkflow: 技术博客质量评估工作流 def __init__(self): self.fetcher ContentFetcher() self.extractor InformationExtractor() self.evaluator MultiDimensionEvaluator() self.reporter ReportGenerator() self.logger logging.getLogger(__name__) def run(self, url: str) - BlogEvaluationReport: 执行完整的工作流 self.logger.info(fStarting Blog Evaluation Workflow for URL: {url}) # 步骤 1: 初始化数据对象 blog_data BlogContent(urlurl, title) # 步骤 2: 抓取内容 self.logger.info(Step 1/4: Fetching content...) blog_data self.fetcher.run(blog_data) # 步骤 3: 提取信息 self.logger.info(Step 2/4: Extracting information...) blog_data self.extractor.run(blog_data) # 步骤 4: 多维度评估 self.logger.info(Step 3/4: Evaluating dimensions...) evaluation_results self.evaluator.run(blog_data) # 步骤 5: 生成报告 self.logger.info(Step 4/4: Generating final report...) final_report self.reporter.run(blog_data, evaluation_results) self.logger.info(Blog Evaluation Workflow completed successfully.) return final_report3.4 创建 API 服务入口 (app/main.py)最后我们使用 FastAPI 将工作流包装成一个 HTTP 服务使其可以被外部系统调用。from fastapi import FastAPI, HTTPException from pydantic import BaseModel, HttpUrl from app.workflows.blog_evaluator import BlogEvaluationWorkflow import logging import uvicorn # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app FastAPI(titleAI Blueprint Demo - Blog Evaluator API, version1.0.0) # 初始化工作流可考虑使用依赖注入或单例优化 workflow BlogEvaluationWorkflow() class EvaluationRequest(BaseModel): url: HttpUrl # 未来可扩展其他参数如评估维度自定义、模型选择等 app.post(/evaluate, summary评估一篇技术博客的质量) async def evaluate_blog(request: EvaluationRequest): 接收一篇技术博客的URL启动评估工作流返回结构化报告。 try: logger.info(fReceived evaluation request for URL: {request.url}) report workflow.run(str(request.url)) return { success: True, data: report.dict() # 将Pydantic模型转为字典 } except Exception as e: logger.exception(fWorkflow execution failed for {request.url}) raise HTTPException(status_code500, detailfInternal workflow error: {str(e)}) app.get(/health) async def health_check(): return {status: healthy} if __name__ __main__: # 用于本地开发运行 uvicorn.run(app.main:app, host0.0.0.0, port8000, reloadTrue)4. 运行验证与结果分析4.1 配置与启动服务首先在项目根目录创建.env文件配置你的 OpenAI API Key。# .env OPENAI_API_KEYsk-your-openai-api-key-here然后启动 FastAPI 服务。cd ai_blueprint_demo python -m app.main服务将在http://localhost:8000启动。你可以访问http://localhost:8000/docs查看自动生成的 API 文档Swagger UI。4.2 调用 API 进行测试使用curl或任何 API 测试工具如 Postman调用接口。curl -X POST \ http://localhost:8000/evaluate \ -H Content-Type: application/json \ -d { url: https://example.com/your-tech-blog-post }请将https://example.com/your-tech-blog-post替换为一篇真实的技术博客文章地址。4.3 解读输出结果一个成功的响应将返回一个结构化的 JSON 报告其结构遵循我们定义的BlogEvaluationReport模型。示例如下{ success: true, data: { blog: { url: https://example.com/your-tech-blog-post, title: 深入理解Python异步编程, cleaned_text: ...清理后的Markdown正文..., code_snippets: [import asyncio\n..., async def main():...], metadata: {} }, overall_score: 7.8, dimension_scores: [ { dimension: 可读性, score: 8, comment: 文章结构清晰语言流畅但部分段落稍显冗长。 }, { dimension: 技术深度, score: 9, comment: 对asyncio的事件循环和协程原理剖析到位。 }, // ... 其他维度 ], summary: 这是一篇关于Python异步编程的优质文章原理讲解透彻代码示例实用。, suggestions: [ 可在文章开头增加一个‘快速开始’的迷你示例降低入门门槛。, 关于性能对比的数据可以更可视化例如增加图表。, 可以补充一些常见的异步编程‘坑’与调试技巧。 ] } }这个输出清晰地展示了工作流的成果原始 URL 经过四个步骤转化为了一个包含量化评分、维度评语、总结和改进建议的完整报告。每个步骤的输入输出都被严格定义整个流程可观测、可调试。5. 常见问题排查与工程化建议将 AI 工作流工程化必然会遇到各种问题。以下是基于此项目的常见排查点和优化建议。5.1 工作流执行失败排查清单当 API 调用失败或返回异常时可按此清单逐步排查。问题现象可能原因检查方式处理建议启动服务时报ModuleNotFoundError依赖未安装或虚拟环境未激活运行pip list | grep -E (fastapi|langchain)确认虚拟环境已激活并重新安装依赖pip install -r requirements.txt调用/evaluate返回500错误日志显示OpenAI API错误API Key 未配置或无效网络问题1. 检查.env文件是否存在且OPENAI_API_KEY正确。2. 尝试用curl直接调用 OpenAI API 测试。1. 修正.env文件。2. 检查网络连接和防火墙设置。3. 确认 API 额度充足。工作流卡在“抓取内容”步骤超时目标网站访问慢、需要 JS 渲染、或触发了反爬1. 查看日志中 Playwright 的报错信息。2. 手动用浏览器访问该 URL 测试。1. 增加ContentFetcher的timeout_ms参数。2. 考虑添加 User-Agent 等请求头。3. 对于复杂页面可尝试使用page.wait_for_selector等待特定元素。评估结果分数全部为5分或评语异常LLM 输出解析失败降级逻辑生效查看evaluator.py中_evaluate_single_dimension方法的日志检查 LLM 原始响应。1. 优化提示词 (evaluation_prompt)要求 LLM 输出更规范的 JSON。2. 增强PydanticOutputParser的错误处理和重试机制。3. 使用 LangChain 的RetryOutputParser。报告中的“总结”或“建议”字段为空ReportGenerator对 LLM 输出的解析规则过于简单打印ReportGenerator.run方法中的llm_output变量观察实际返回文本格式。1. 使用更强大的解析库如guardrails-ai或instructor。2. 改用PydanticOutputParser来解析整个报告而不仅仅是单个评估结果。5.2 将工作流推向生产环境的建议上述示例是一个用于学习和演示的“最小可行产品”MVP。要用于生产环境还需要考虑以下方面配置管理将模型类型、API 地址、超时时间、重试次数等参数外置到配置文件如config.yaml或环境变量中避免硬编码。异步处理博客评估可能耗时较长数十秒FastAPI 的同步端点会阻塞。应改为异步端点并使用 Celery、RQ 或 FastAPI 的BackgroundTasks进行异步任务处理立即返回一个任务 ID客户端通过轮询或 WebSocket 获取结果。持久化与状态管理将工作流执行状态、中间结果和最终报告存入数据库如 PostgreSQL、Redis便于查询、重试和审计。可观测性集成结构化日志如 JSON 格式并接入日志收集系统如 ELK。为关键节点添加指标如耗时、成功率使用 Prometheus 和 Grafana 进行监控。节点容错与降级抓取失败可以配置备用抓取方式如直接requests获取静态 HTML或对特定网站使用不同的解析策略。LLM 调用失败实现指数退避重试机制。对于非核心评估维度允许部分失败。工作流编排引擎对于更复杂、分支众多的工作流可以考虑使用专门的编排引擎如 Apache Airflow、Prefect 或 Dagster它们提供了更强大的调度、依赖管理、可视化界面和错误处理能力。版本化与回滚工作流定义即蓝图应进行版本控制。当新版本的工作流出现问题时能快速回滚到旧版本。6. 扩展方向与最佳实践基于这个“博客评估”蓝图你可以将其思想扩展到无数场景。6.1 扩展场景示例AI 客服工单处理工作流节点包括“用户意图识别”、“知识库检索”、“答案生成”、“安全审核”、“多轮对话管理”。内容审核流水线节点包括“文本敏感词检测”、“图片 OCR 与违规识别”、“AI 多模态综合判断”、“人工复核队列分发”、“结果同步与日志”。数据分析报告生成节点包括“连接数据源执行 SQL”、“数据清洗与转换”、“关键指标计算”、“图表生成”、“洞察文本生成”、“报告排版与导出”。6.2 工程师式 AI 工作流设计最佳实践单一职责每个节点只做一件事并把它做好。这降低了复杂度便于测试和复用。强类型接口使用像 Pydantic 这样的工具严格定义节点间传递的数据模型。这是避免“字符串地狱”和运行时错误的关键。无状态设计尽可能让节点是无状态的其输出完全由输入决定。这使节点易于测试、并行化和缓存。明确的错误边界每个节点都应定义清楚可能发生的错误并在节点内部或工作流层面设计处理策略重试、降级、快速失败。配置驱动将模型参数、API 端点、业务规则等作为配置使工作流能适应不同环境开发、测试、生产和客户需求而无需修改代码。蓝图即代码将工作流的结构节点、连接关系也用代码或声明式配置如 YAML来定义并将其纳入版本控制系统。这是实现 CI/CD 和团队协作的基础。回到开头的问题工程师式 AI 工作流的核心价值在于它将 AI 能力从“黑盒咒语”变成了“白盒蓝图”。你不再需要反复调试一个巨大的、模糊的提示词而是可以像调试普通程序一样设置断点、查看中间变量、替换某个故障模块。当需求变化时你可以通过增删节点、调整连接来快速响应而不是重写整个提示词。这种可维护性、可观测性和可复用性正是 AI 应用从玩具走向生产系统所必需的工程基石。