前言：Agent 能跑 Demo，但能跑生产吗？
前几周我们写了手搓 Agent 系列——从 ReAct 到 MCP，技术栈完整。但技术能跑 Demo 是一个问题，能跑生产是另一个问题。
最近三个真实案例值得关注：
- 首钢：热轧产线计划单审核从 1 小时缩到秒级，安全隐患识别效率提升 60%
- 京东云：通过推理优化，同样业务从 2700 元降到 10 元
- CSDN 创始人：预判未来一两年 99% 的代码可能由 AI 智能体生成
这篇文章讲的是 Demo 到生产的鸿沟怎么跨。
一、Demo 和生产之间差了什么？
| 维度 | Demo 阶段 | 生产阶段 |
|---|---|---|
| 任务确定性 | "帮我分析这段代码" | "分析 PR #342 的安全性，只关注 SQL 注入和 XSS，输出 JSON 格式" |
| 错误处理 | 打印异常信息 | 自动重试→降级→告警→人工介入 |
| 成本 | 不管 | 每次调用追踪到分，日预算硬上限 |
| 可观测性 | print 语句 | 结构化日志 + 链路追踪 + 成本仪表盘 |
| 人机协作 | 全自动 | 高风险操作人工审批，低风险自动 |
| 集成方式 | 命令行 | REST API / Webhook / 企微/钉钉/飞书 |
二、任务拆分：不要让 Agent 一口吃成胖子
生产环境中最常见的错误：给 Agent 一个模糊的大任务，期待它自己搞定一切。
错误示范
"帮我优化这个系统的性能"
Agent 不知道从哪里开始，可能会：读取所有文件 → 分析所有代码 → 给出 5000 字的建议 → Token 超限 → 任务失败。
正确做法：任务原子化
class TaskDecomposer:
"""将大任务拆分为 Agent 可执行的原子任务"""
def decompose(self, task: str) -> list[dict]:
"""LLM 辅助拆分 + 人工确认"""
prompt = f"""将以下任务拆分为 Agent 可执行的原子步骤。
每个步骤应：单一职责、有明确输入输出、可独立验证。
任务：{task}
输出 JSON 数组，每项包含：
- id: 步骤编号
- title: 简短描述
- input: 需要的输入数据
- output: 期望的输出
- tool: 需要的工具
- risk: low/medium/high
"""
# 实际调用 LLM 获取拆分结果
steps = self._call_llm(prompt)
# 高风险步骤标记需要人工确认
for s in steps:
if s["risk"] == "high":
s["require_approval"] = True
return steps
以"优化系统性能"为例，正确拆分为：
Step 1: 获取慢查询日志（工具：read_log, 风险：low）
Step 2: 分析TOP 5 慢查询（工具：analyze_sql, 风险：low）
Step 3: 为每个慢查询生成优化SQL（工具：generate_sql, 风险：medium）
Step 4: 检查优化SQL是否安全（工具：validate_sql, 风险：high，需确认）
Step 5: 输出优化报告（工具：write_report, 风险：low）
三、成本控制：从 2700 到 10 元
京东云的实际案例：同样的业务逻辑，通过以下优化把成本从 2700 元降到 10 元。
| 优化手段 | 节省 | 原理 |
|---|---|---|
| 模型降级 | 60% | 80% 的请求不需要旗舰模型 |
| 缓存复用 | 25% | 相似问题直接返回缓存结果 |
| Prompt 精简 | 10% | 去掉冗余指令，Token 减半 |
| 批量合并 | 5% | 多个小请求合并为一次大请求 |
class ProductionCostController:
"""生产成本控制器"""
def __init__(self, daily_budget: float = 50.0):
self.daily_budget = daily_budget
self.spent_today = 0.0
self.cache = {}
def before_call(self, task: str, complexity: str) -> str:
"""调用前检查：预算 + 缓存 + 模型选择"""
# 1. 预算检查
if self.spent_today >= self.daily_budget:
return "BUDGET_EXCEEDED"
# 2. 缓存命中
cache_key = hashlib.md5(task.encode()).hexdigest()
if cache_key in self.cache:
return "CACHE_HIT"
# 3. 模型选择
if complexity == "low":
return "luna" # $1/1M
elif complexity == "medium":
return "terra" # $2.5/1M
else:
return "sol" # $5/1M
def after_call(self, task: str, result: str, cost: float):
"""调用后：记录成本 + 缓存结果"""
self.spent_today += cost
cache_key = hashlib.md5(task.encode()).hexdigest()
self.cache[cache_key] = {"result": result, "cost": cost, "time": time.time()}
四、人机协作：不要全自动
生产环境中，Agent 应该能区分"可以自己做"和"需要问人"。
```python
class HumanInTheLoop:
"""人机协作决策器"""
AUTO_APPROVE_RISK = ["low"]
REQUIRE_APPROVAL_RISK = ["high", "critical"]
ASK_IF_UNCERTAIN_RISK = ["medium"]
def decide(self, action: dict, confidence: float) -> str:
"""返回: auto / ask / block"""
risk = action.get("risk", "medium")
if risk in self.AUTO_APPROVE_RISK and confidence > 0.8:
return "auto"
if risk in self.REQUIRE_APPROVAL_RISK:
return "block" # 必须人工审批
if risk in self.ASK_IF_UNCERTAIN_RISK and confidence 如果觉得有用，欢迎 **点赞 + 收藏 + 关注**。手搓 Agent 系列已完结，本文是生产落地篇的补充。
🔜 下午：AI 编程工具市场格局——ChatGPT 份额跌破 50% 后的生态变化