AI Agent生产落地实战：可执行、可调试、可交付的工作流设计-平芜编程栈

1. 这不是概念炒作，是我在真实项目里反复拆解、重装、推翻重来的血泪笔记

“AI Agent 到底是什么？”——这句话我去年在 Slack 群里问过 7 次，每次得到的答案都不一样：有人说是“带记忆的 ChatGPT”，有人说是“能自动点鼠标的应用”，还有人直接甩来一张画满箭头的架构图，标注着 “LLM + Tool Use + Planning + Memory”。我信了，回去就照着搭，结果花 3 天配好环境，第 4 天发现它连“帮我把邮箱里上周五收到的 PDF 报告转成 Excel 表格”这种任务都卡死在“思考下一步该调哪个 API”上，最后报错停在ToolNotFoundException: find_excel_converter。那一刻我才明白：市面上 80% 的“Agent 教程”，讲的都不是怎么让 Agent 跑起来，而是怎么让 PPT 动起来。

我用整整一年时间，从零开始跑通了 12 个不同复杂度的 Agent 实际场景：自动整理会议纪要并生成待办清单、跨平台同步客户信息（飞书+CRM+本地 Excel）、根据销售日报自动生成周报 PPT、监控竞品官网价格变动并触发邮件预警、甚至用语音指令控制树莓派小车绕开障碍物……前后烧掉硬件成本 2360 元（含两块 Jetson Nano、三台旧 Mac Mini 做分布式推理节点、云服务账单），调试日志文件攒了 47GB。这篇不是教科书定义，也不是厂商白皮书，是我把每个模块拆到焊点级别、把每条报错日志反向追踪到源码行号、把每个“看起来很酷”的功能拉回现实场景反复验证后，写给真正想动手做点东西的人的实操手记。如果你刚看完某篇“5 分钟打造你的专属 Agent”，却发现连pip install crewai都报错缺 C++ 编译器；或者你已经能跑通 demo，但一加真实业务逻辑就崩，那这篇就是为你写的。它不讲“智能体范式演进”，只讲“为什么你改了 system prompt 还是调不动工具”；不谈“多智能体社会涌现”，只说“怎么让两个 Agent 在共享数据库里不互相覆盖数据”。核心关键词就三个：可执行、可调试、可交付——不是 Demo 可运行，是下周就要上线给老板看的版本，能扛住 200 条并发请求、日志里能精准定位到哪一行代码让 Agent 把“张经理”错认成“章经理”。

2. 内容整体设计与思路拆解：放弃“通用智能体”，拥抱“场景专用工作流”

2.1 为什么所有“全能型 Agent 框架”在真实业务中都容易翻车？

我最早试的是 LangChain 的AgentExecutor，配置了OpenAIFunctionsAgent，绑了几个工具：查天气、搜维基、发邮件。跑 demo 很炫——输入“告诉我北京今天会不会下雨，如果会就给王总发邮件提醒带伞”。它真去查了天气，也真发了邮件。但当我把“查天气”换成“从公司 CRM 导出近 30 天成交客户名单”，问题立刻暴露：CRM 接口返回的是分页 JSON，每页 50 条，而 Agent 的tool_input解析器默认只处理单层字符串，根本没法传{"page": 1, "per_page": 50}这种结构化参数。更致命的是，它不会主动判断“导出名单”需要分页循环，也不会在第一次调用失败（比如 token 超限）后降级重试。它只会卡住，然后输出一句：“我无法完成该任务。”

后来我转向 AutoGen，用GroupChatManager拉起三个角色：Product Manager、Engineer、Reviewer。设定好角色描述和工具，让它“评估新功能上线风险”。结果它真开了个“会议”，三个 Agent 轮流发言，最后 Engineer 输出了一段技术可行性分析，Reviewer 回复“同意”，Product Manager 总结“风险可控”。听起来很专业？但所有分析都基于它自己编造的“假设数据”，没人告诉它去读 Jira 里的 PR 描述、没人让它调用 SonarQube API 查代码质量报告、更没人限制它不能把“内存泄漏”误判为“UI 布局问题”。这根本不是协作，是三个演员在即兴表演。

根本症结在于：这些框架预设了一个“理想世界”——工具永远可用、API 永远返回标准格式、用户指令永远清晰无歧义、错误永远可被 LLM 自动恢复。而现实是：CRM 接口凌晨三点维护、Excel 文件里混着合并单元格和空行、销售同事发来的需求是“把那个蓝色表格弄成能筛选的”，连字段名都没提。

所以我的设计思路彻底转向：不追求“一个 Agent 通吃所有事”，而是为每个具体业务动作，定制一个最小可行工作流（Minimal Viable Workflow, MVW）。比如“自动整理会议纪要”这个需求，我不再试图让一个 Agent 理解全部语音转文字、提取要点、生成待办、同步到飞书日历——而是拆成四个原子步骤：

语音转文字（ASR）：用 Whisper.cpp 在本地 Mac Mini 上离线运行，输出带时间戳的 SRT；
关键信息抽取（IE）：用微调过的 tiny-llama-1.1b 模型，在 Jetson Nano 上跑，只识别“决策项”、“负责人”、“截止时间”三类实体；
待办生成（TO-DO Gen）：用规则引擎（Drools）匹配抽取结果，生成标准化待办 JSON（含assignee,due_date,description字段）；
飞书同步（Sync）：调用飞书开放平台 API，将 JSON 写入指定多维表格。

每个环节都是独立进程，有明确输入/输出契约，失败时能精准上报错误类型（如ASR_TIMEOUT,IE_MODEL_OOM），上游可按策略重试或降级（如 ASR 失败则启用备用语音服务商；IE 模型 OOM 则切回 CPU 模式降速运行）。整个流程用 Airflow 编排，可视化 DAG 图里每个节点状态一目了然。这不是“智能”，这是把 AI 当作一个可插拔的、带一定容错能力的组件，嵌入到已有的工程化流水线里。

2.2 为什么我坚持用“轻量模型+强规则”组合，而不是全靠大模型？

很多人看到“Agent”第一反应就是上 GPT-4 或 Claude 3。我试过——用 GPT-4-turbo 处理销售日报生成周报 PPT。输入是 500 字纯文本日报，输出是 Markdown 格式，再用 python-pptx 渲染。效果惊艳：标题大气、逻辑清晰、甚至加了“建议关注”小节。但问题接踵而至：

成本不可控：一份日报平均消耗 1200 tokens，按 $0.01/1K tokens 算，每月 200 份就是 $2.4，看似不多。但当销售总监要求“把上季度所有日报汇总分析”，token 消耗瞬间飙升到 8 万，单次成本 $80，且响应时间从 3 秒涨到 47 秒；
输出不稳定：同一份日报，连续调用 5 次，生成的 PPT 结构有 3 种变体，其中一次把“Q3 销售额增长 12%”错写成“Q3 销售额下降 12%”，因为模型把负号识别成了减号；
调试黑洞：当输出错误时，你无法知道是 prompt 写得不够好，还是模型本身在特定上下文下产生了幻觉，抑或是输入文本里有个隐藏的 Unicode 字符干扰了 tokenization。

于是我转向“混合模式”：用小模型（<1B 参数）做确定性任务，用大模型（GPT-4/Claude）做模糊性决策。具体到周报生成：

小模型负责“硬解析”：用微调的 distilbert-base-uncased，在本地 GPU 上跑，专门识别日报中的数字（销售额、增长率、客户数）、人名（张经理、李总监）、时间节点（“上周”、“本月底”、“Q3”）。它输出结构化 JSON：{"revenue": 1250000, "growth_rate": 0.12, "key_person": "张伟", "deadline": "2024-09-30"}。这个过程稳定、快速、可复现，错误率 <0.3%；
大模型只负责“软包装”：把结构化 JSON 和固定模板（含公司品牌色、字体规范）喂给 GPT-4，指令极其简单：“请将以下数据填入模板，保持原有格式，不要添加任何额外分析或建议。” 它不再需要“理解”业务，只是个高级填充工具。token 消耗降到 200 以内，响应时间稳定在 1.8 秒，且输出 100% 一致。

这个思路的核心逻辑是：把 AI 的“不可控创造性”关进笼子，只让它做最擅长的“模式匹配”和“文本生成”，而把“必须 100% 准确”的部分，交给经过充分测试的小模型或传统规则引擎。就像汽车里的 ABS 系统——它不决定你往哪开（那是司机的事），但它确保你在急刹时轮胎不抱死（这是确定性保障）。

2.3 为什么我把“记忆”从“向量库”降级为“结构化数据库”？

几乎所有 Agent 教程都会强调“长期记忆”，标配方案是 ChromaDB + OpenAI Embeddings。我搭过一套：会议录音转文字存进 Chroma，用户问“上次提到的服务器迁移方案，张经理怎么说的？”，Agent 去向量检索，找到相关片段，再让 LLM 总结。理论上很美。实际运行一周后，我发现了三个硬伤：

语义漂移严重：用户问“张经理对迁移方案的态度”，向量检索可能召回“张经理说服务器机柜要换新”的片段，但原文其实是“张经理反对换新，认为现有设备还能用三年”。因为“换新”和“反对”在向量空间里距离很近，而“反对”和“态度”却可能很远；
上下文丢失：检索到的片段是孤立的句子，缺乏原始对话的语气、停顿、前后逻辑。LLM 总结时容易断章取义；
维护成本高：Embedding 模型升级（如从 text-embedding-ada-002 换到 text-embedding-3-small），所有历史数据都要重新向量化，10 万条记录耗时 8 小时，期间服务不可用。

我的解决方案是：放弃“语义记忆”，回归“事实记忆”——所有关键信息，强制结构化入库。以会议纪要为例：

ASR 输出 SRT 后，IE 模型抽取的每一条“决策项”，都写入 PostgreSQL 表meeting_decisions，字段包括meeting_id,decision_text,assignee,due_date,status（待办/进行中/已完成）；
每一条“风险点”，写入meeting_risks表，字段含risk_description,owner,mitigation_plan,probability,impact；
所有发言者姓名、角色、所属部门，从飞书日历事件中自动同步到meeting_participants表。

当用户问“张经理负责的待办事项有哪些？”，系统直接 SQL 查询：

SELECT decision_text, due_date, status FROM meeting_decisions WHERE assignee = '张伟' AND status != '已完成';

结果 100% 准确、毫秒级返回、无需 LLM 干预。而真正的“记忆增强”只用在极少数场景：比如用户问“为什么张经理一直反对服务器迁移？”，这时才用向量库检索所有含“张经理”和“服务器迁移”的历史会议片段，作为 LLM 的补充背景，但它不再是唯一答案来源，而是辅助证据。

这背后的理念是：在业务系统里，“准确”比“智能”重要一百倍。一个能“理解”但经常出错的 Agent，不如一个“不懂”但永远正确的数据库查询。

3. 核心细节解析与实操要点：从 Prompt 工程到工具链打磨

3.1 Prompt 不是魔法咒语，是给 LLM 的“操作手册”

很多人以为写好 Prompt 就万事大吉。我踩过最深的坑是：用同一个 Prompt，在 GPT-3.5 和 GPT-4 上表现天差地别。比如这个经典指令：

“你是一个专业的销售助理，请根据以下会议记录，提取所有待办事项，格式为：- [负责人] 事项描述（截止日期）”

在 GPT-3.5 上，它能稳定输出：

[张伟] 跟进客户 A 的合同续签（2024-09-30）
[李娜] 准备 Q4 产品演示材料（2024-10-15）

但在 GPT-4 上，它突然开始发挥：“- [张伟] 作为客户关系维护专家，需深度跟进客户 A 的合同续签事宜，确保在 2024 年 9 月 30 日前完成所有法律审核流程……” —— 这完全违背了“简洁、结构化”的核心要求。

根本原因在于：大模型越强，其“自由发挥欲”越旺盛。GPT-4 认为自己“应该”提供更详尽的服务，而忽略了你的格式约束。解决方法不是骂模型“不听话”，而是把它当成一个需要精确指令的实习生：

明确角色边界：把“专业销售助理”改成“销售助理（严格遵循以下格式输出，禁止任何解释、禁止任何额外内容）”；
强制格式锚点：在 Prompt 开头和结尾都加上分隔符，并规定“输出必须严格位于---OUTPUT START---和---OUTPUT END---之间，且仅包含待办列表”；
提供正例+反例：给出 1 个正确输出示例，再给出 1 个典型错误示例（如带解释、带编号、漏截止日期），并注明“这是错误的，禁止模仿”；
设置温度（temperature）为 0：彻底关闭随机性，确保相同输入必得相同输出。

最终稳定的 Prompt 片段如下（已用于生产环境 6 个月）：

---ROLE--- 你是一个销售助理，职责是严格、准确、简洁地提取待办事项。你不会解释、不会总结、不会添加任何未提及的信息。你的输出必须100%符合下方格式要求。 ---FORMAT RULES--- - 每条待办以短横线开头，后跟空格 - 格式为：[负责人] 事项描述（YYYY-MM-DD） - 负责人必须是原文中出现的完整姓名（如“张伟”，非“张经理”） - 事项描述必须是原文原意，禁止扩写、禁止推测 - 截止日期必须是原文中明确写出的日期，格式为 YYYY-MM-DD；若原文未写，则此项留空，不猜测 ---EXAMPLE CORRECT--- ---OUTPUT START--- - [张伟] 跟进客户A的合同续签（2024-09-30） - [李娜] 准备Q4产品演示材料（2024-10-15） ---OUTPUT END--- ---EXAMPLE WRONG--- - [张伟] 作为客户关系维护专家，需深度跟进客户A的合同续签事宜（2024-09-30） // 错误：添加了身份描述和修饰词 - 1. [张伟] 跟进客户A的合同续签（2024-09-30） // 错误：添加了序号 ---INPUT--- {meeting_transcript} ---OUTPUT START---

提示：这个 Prompt 在 GPT-3.5、GPT-4、Claude 3、甚至本地 llama3-8b 上都能保持 99.2% 的格式一致性。关键不是“多聪明”，而是“多强硬”。把 LLM 当成一台需要精确 G 代码的 CNC 机床，而不是一个可以聊家常的朋友。

3.2 工具（Tools）不是插件，是必须签订“服务协议”的外部系统

教程里常说“给 Agent 加个工具很简单，几行代码就行”。比如 LangChain 的Tool类：

def search_weather(city: str) -> str: return requests.get(f"https://api.weather.com/v3/weather/forecast?city={city}").json()["temp"] weather_tool = Tool( name="search_weather", func=search_weather, description="Useful for getting current weather in a city" )

这在 demo 里没问题。但放到生产环境，问题立刻爆发：

超时无感知：天气 API 响应慢于 5 秒，Agent 就卡死，不会自动重试或降级；
错误不分类：API 返回 404（城市不存在）和 503（服务不可用），Agent 都当成“工具调用失败”，无法区分是用户输错城市，还是服务宕机；
输入无校验：用户输入search_weather("北京&script>alert('xss')")，函数直接拼接 URL，引发安全漏洞。

我的工具封装原则是：每个工具必须自带“健康检查”、“熔断机制”、“输入净化”和“错误分类”。以 CRM 数据导出工具为例：

class CRMExportTool: def __init__(self, api_client): self.client = api_client # 熔断器：连续3次失败，暂停10分钟 self.circuit_breaker = CircuitBreaker(failure_threshold=3, timeout=600) def _validate_input(self, params: dict) -> dict: """输入净化与校验""" # 清洗SQL注入字符 clean_params = {k: re.sub(r"[;'\-\+\*]", "", str(v)) for k, v in params.items()} # 强制分页参数 clean_params["page"] = max(1, min(100, int(clean_params.get("page", 1)))) clean_params["per_page"] = max(10, min(100, int(clean_params.get("per_page", 50)))) return clean_params def _handle_error(self, e: Exception) -> ToolError: """错误分类""" if isinstance(e, requests.Timeout): return ToolError(code="TIMEOUT", message="CRM服务响应超时，请稍后重试") elif isinstance(e, requests.ConnectionError): return ToolError(code="CONNECTION_FAILED", message="无法连接CRM服务，请检查网络") elif hasattr(e, "response") and e.response.status_code == 401: return ToolError(code="AUTH_EXPIRED", message="CRM登录凭证已过期，请联系管理员") else: return ToolError(code="UNKNOWN_ERROR", message=f"未知错误：{str(e)}") def run(self, params: dict) -> dict: try: self.circuit_breaker.check() clean_params = self._validate_input(params) response = self.client.get("/api/v1/customers", params=clean_params, timeout=8) response.raise_for_status() return {"data": response.json(), "total_count": response.headers.get("X-Total-Count", 0)} except Exception as e: error = self._handle_error(e) self.circuit_breaker.record_failure() raise error

注意：这个工具类返回的不是原始 JSON，而是一个带code和message的ToolError对象。Agent 的执行层（Executor）会捕获这个异常，并根据code做不同处理：TIMEOUT触发重试；AUTH_EXPIRED则直接中断流程并通知管理员；UNKNOWN_ERROR记录详细日志供排查。这才是生产级工具该有的样子——它不是一个“能用就行”的函数，而是一个有 SLA、有监控、有兜底的微服务。

3.3 “规划”（Planning）不是让 LLM 自由发挥，而是用状态机驱动

很多 Agent 框架的“Planning”模块，本质是让 LLM 生成一段自然语言描述的执行步骤，比如：

“第一步：调用 search_weather 获取北京天气；第二步：调用 send_email 将结果发送给张经理”

这在简单场景下可行。但一旦涉及分支逻辑（“如果天气 >30°C，就调用空调控制 API；否则调用加湿器 API”），LLM 生成的计划就变得不可靠——它可能漏掉条件判断，可能把“>30°C”错写成“>=30°C”，甚至可能生成根本不存在的工具名。

我的方案是：用有限状态机（FSM）替代 LLM 规划。为每个业务流程定义明确的状态和转移规则。以“客户跟进自动化”为例，定义状态：

WAITING_FOR_FIRST_CONTACT（等待首次联系）
FOLLOW_UP_SENT（已发送跟进邮件）
MEETING_SCHEDULED（已预约会议）
DEAL_CLOSED（成交）
DEAL_LOST（丢单）

转移规则由硬编码逻辑控制：

class CustomerFollowUpFSM: def __init__(self, customer_id): self.state = "WAITING_FOR_FIRST_CONTACT" self.customer_id = customer_id def on_email_sent(self): if self.state == "WAITING_FOR_FIRST_CONTACT": self.state = "FOLLOW_UP_SENT" # 触发：设置 3 天后自动检查邮件回复 schedule_job("check_email_reply", self.customer_id, delay=3*24*3600) def on_meeting_confirmed(self): if self.state in ["FOLLOW_UP_SENT", "WAITING_FOR_FIRST_CONTACT"]: self.state = "MEETING_SCHEDULED" # 触发：生成会议议程，发送日历邀请 def on_deal_closed(self): if self.state == "MEETING_SCHEDULED": self.state = "DEAL_CLOSED" # 触发：生成合同，通知法务

LLM 在这里只承担一个角色：状态识别器（State Classifier）。当收到一封新邮件，系统不问“接下来该做什么”，而是问 LLM：“请判断这封邮件属于以下哪个状态：WAITING_FOR_FIRST_CONTACT, FOLLOW_UP_SENT, MEETING_SCHEDULED, DEAL_CLOSED, DEAL_LOST。只输出状态名，不加任何其他内容。” 它的任务被极度简化，准确率从规划模式的 72% 提升到 98.5%。

实操心得：把“规划权”交给状态机，把“判断权”交给 LLM，各司其职。就像交规——红灯停、绿灯行是铁律（状态机），而“前面那辆车是不是在玩手机”这种模糊判断，才交给驾驶员（LLM）。

4. 实操过程与核心环节实现：从零搭建一个可交付的销售日报生成 Agent

4.1 环境准备与依赖安装：避开 Python 包地狱

我用的是 Ubuntu 22.04 LTS（服务器） + macOS 14（开发机）双环境。核心原则：所有依赖必须锁定版本，所有环境必须可重现。绝不接受pip install langchain这种写法。

Python 环境：用pyenv管理，项目固定使用3.11.6。原因：3.12+的某些异步特性与旧版httpx冲突，而langchain0.1.x 系列尚未完全适配；

包管理：不用requirements.txt，改用poetry。pyproject.toml中明确声明：

[tool.poetry.dependencies] python = "^3.11.6" langchain = { version = "^0.1.16", allow-prereleases = true } langchain-openai = "^0.1.3" psycopg2-binary = "^2.9.7" # 必须用 binary，源码编译在 M1 Mac 上会失败 openai = "^1.13.3" # 注意：v1.x API 与 v0.x 完全不兼容

最关键的避坑点：openai包和langchain-openai的版本必须严格匹配。我曾因openai==1.14.0与langchain-openai==0.1.2不匹配，导致OpenAIEmbeddings初始化时报AttributeError: 'OpenAI' object has no attribute 'embeddings'，排查了 7 小时才发现是版本错配。Poetry 的lock文件能完美解决此问题。

硬件方面，Jetson Nano（4GB 版）跑tiny-llama-1.1b是够用的，但必须关闭所有 GUI 进程，只留systemd和nginx，否则内存不足会触发 OOM Killer 杀掉模型进程。我写了个启动脚本/etc/systemd/system/llm-inference.service：

[Unit] Description=LLM Inference Service After=network.target [Service] Type=simple User=llm WorkingDirectory=/opt/llm-inference ExecStart=/usr/bin/python3 -m vllm.entrypoints.api_server --model /opt/models/tiny-llama-1.1b --tensor-parallel-size 1 --gpu-memory-utilization 0.85 --host 0.0.0.0 --port 8000 Restart=always RestartSec=10 MemoryLimit=3G OOMScoreAdjust=-900 [Install] WantedBy=multi-user.target

OOMScoreAdjust=-900是关键——告诉 Linux 内核，这个进程绝不能被 OOM Killer 杀掉，宁可杀其他进程。MemoryLimit=3G则硬性限制其内存使用，防止它吃光所有 RAM。

4.2 核心工作流编排：Airflow DAG 的实战配置

我放弃LangChain的AgentExecutor，选择 Apache Airflow 作为编排引擎。理由：可视化 DAG、失败重试、依赖管理、权限控制、审计日志，全是生产必需。以下是销售日报生成 DAG 的核心代码（dags/sales_report_dag.py）：

from airflow import DAG from airflow.operators.python import PythonOperator from airflow.providers.postgres.operators.postgres import PostgresOperator from airflow.models import Variable from datetime import datetime, timedelta import json # 从 Airflow Variables 读取敏感配置，避免硬编码 OPENAI_API_KEY = Variable.get("openai_api_key") CRM_API_URL = Variable.get("crm_api_url") default_args = { 'owner': 'ai-team', 'depends_on_past': False, 'start_date': datetime(2024, 1, 1), 'email_on_failure': True, 'email': ['ops@company.com'], 'retries': 3, # 每个任务失败自动重试3次 'retry_delay': timedelta(minutes=5), # 重试间隔5分钟 } dag = DAG( 'sales_daily_report', default_args=default_args, description='Generate daily sales report from CRM data', schedule_interval='0 8 * * *', # 每天早上8点执行 catchup=False, tags=['sales', 'ai', 'report'], ) def fetch_crm_data(**context): """Step 1: 从CRM拉取昨日销售数据""" from utils.crm_client import CRMClient client = CRMClient(CRM_API_URL) # Airflow 的 XCom 机制，把结果传给下游 context['task_instance'].xcom_push( key='crm_data', value=client.get_yesterday_sales() ) def extract_metrics(**context): """Step 2: 用小模型提取关键指标""" from utils.llm_extractor import TinyLlamaExtractor extractor = TinyLlamaExtractor(model_path="/opt/models/tiny-llama-1.1b") crm_data = context['task_instance'].xcom_pull(key='crm_data') metrics = extractor.extract(crm_data) context['task_instance'].xcom_push(key='metrics', value=metrics) def generate_report_md(**context): """Step 3: 用GPT-4生成Markdown报告""" from openai import OpenAI client = OpenAI(api_key=OPENAI_API_KEY) metrics = context['task_instance'].xcom_pull(key='metrics') # 构造极简Prompt，只做填充 prompt = f""" 请将以下销售数据填入报告模板，保持原有格式，不要添加任何分析或建议： 销售额：{metrics['revenue']} 新增客户数：{metrics['new_customers']} 重点跟进客户：{', '.join(metrics['key_accounts'])} 模板： # 销售日报 - {datetime.now().strftime('%Y-%m-%d')} ## 核心指标 - 销售额：{{revenue}} - 新增客户数：{{new_customers}} ## 重点客户 {{key_accounts}} """ response = client.chat.completions.create( model="gpt-4-turbo", messages=[{"role": "user", "content": prompt}], temperature=0, max_tokens=500 ) report_md = response.choices[0].message.content context['task_instance'].xcom_push(key='report_md', value=report_md) def save_to_db(**context): """Step 4: 保存报告到PostgreSQL""" from utils.db import get_db_connection report_md = context['task_instance'].xcom_pull(key='report_md') conn = get_db_connection() with conn.cursor() as cur: cur.execute( "INSERT INTO sales_reports (date, content, generated_at) VALUES (%s, %s, NOW())", (datetime.now().date(), report_md) ) conn.commit() # 定义任务 fetch_task = PythonOperator( task_id='fetch_crm_data', python_callable=fetch_crm_data, dag=dag, ) extract_task = PythonOperator( task_id='extract_metrics', python_callable=extract_metrics, dag=dag, ) generate_task = PythonOperator( task_id='generate_report_md', python_callable=generate_report_md, dag=dag, ) save_task = PythonOperator( task_id='save_to_db', python_callable=save_to_db, dag=dag, ) # 设置依赖关系 fetch_task >> extract_task >> generate_task >> save_task

这个 DAG 的优势在于：每个环节失败，Airflow 都会发邮件告警，并在 Web UI 上清晰标红，点击即可查看完整日志。更重要的是，你可以随时“重放”某一天的流程：在 UI 上选中fetch_crm_data任务，点击“Clear”，它就会重新执行并向下触发整个链条，方便调试。这比在 Jupyter Notebook 里手动跑 10 行代码，可靠一万倍。

4.3 关键参数计算与性能调优：让 Agent 跑得又快又稳

Token 预估与成本控制：GPT-4-turbo 的输入成本是 $0.01/1K tokens，输出是 $0.03/1K tokens。我写了个预估脚本，对每份日报做静态分析：

def estimate_tokens(text: str) -> int: # 粗略估算：英文1 token ≈ 4 chars，中文1 token ≈ 1.3 chars en_chars = len(re.findall(r'[a-zA-Z0-9\s]', text)) zh_chars = len(re.findall(r'[\u4e00-\u9fff]', text)) return int(en_chars / 4 + zh_chars / 1.3) # 在 generate_report_md 任务开头加入 input_tokens = estimate_tokens(prompt) if input_tokens > 8000: # 超过8K，强制截断 prompt = prompt[:int(8000 * 1.3)] + "...（内容已截断）"

这样确保单次调用成本稳定在 $0.15 以内，且不会因超长输入触发 API 限流。

并发控制与速率限制：OpenAI API 有 RPM（每分钟请求数）限制。我用tenacity库实现指数退避：

from tenacity import retry, stop_after_attempt, wait_exponential @retry( stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=4, max=10) ) def call_openai_api(prompt): return client.chat.completions.create(...) # 第一次失败等4秒，第二次等8秒，第三次等10秒（max），最多试5次

本地模型量化与加速：tiny-llama-1.1b原始 FP16 模型约 2.1GB，Jetson Nano 内存只有 4GB。我用llama.cpp的quantize工具将其量化为Q4_K_M格式：
```
./quantize /opt/models/tiny-llama-1.1b/ggml-model-f16.bin /opt/models/tiny-llama-1.1b/ggml-model-Q4_K_M.gguf Q4_K_M
```
量化后模型仅 680MB，推理速度从 3.2 tok/s 提升到 8.7 tok/s，内存占用降至 1.2GB，完全满足实时处理需求。