从需求到上线：AI 原生应用（Agent）的端到端开发实战

很多团队做AI应用时，习惯性地把它当成传统软件来开发：产品经理写PRD，后端画接口文档，前端出原型，研发按部就班实现。但一上生产就发现，这套流程在Agent场景下行不通。

因为大模型是概率性系统。传统软件“输入A必定输出B”，而Agent“输入A可能输出B、C或者一堆乱码”。这种不确定性导致了开发范式的根本改变：从“编写确定性逻辑”变成了“引导和约束概率性推理”。

这篇文章抛开高深架构，以工程落地的视角，完整复盘一个AI原生应用（Agent）从需求分析到上线运维的端到端开发流程。附可运行的代码骨架，希望能帮你避开“Demo惊艳，上线翻车”的坑。

一、 需求阶段：从“功能描述”到“能力边界”

做Agent最忌讳的第一件事，是让产品经理直接写“我要一个能自动分析财报并写投资建议的Agent”。这个需求听起来很酷，但在工程上无法验收。

AI原生的需求分析，核心不是定义“AI能做什么”，而是定义“AI不能做什么”。

我们需要把传统PRD转换为Agent系统契约（Agent Spec）：

# 需求阶段产出的不是PRD，而是结构化的Agent Spec
agent_spec = {
"name": "financial_analyst_agent",
"goal": "基于提供的财报数据生成分析报告",
"capabilities": [
"提取营收、利润、现金流等关键指标",
"对比历史数据计算同比/环比",
"生成风险提示"
],
"constraints": [
"禁止编造未在上下文中提供的财务数据",
"必须强制包含免责声明"
],
"output_schema": {
"type": "object",
"properties": {
"metrics": {"type": "object"},
"risk_alerts": {"type": "array"},
"disclaimer": {"type": "string"}
},
"required": ["metrics", "risk_alerts", "disclaimer"]
}
}

二、 架构设计：从“单体调用”到“Agent Workflow”

在代码层面，不要一上来就追求复杂的多Agent协作。最务实、最易维护的架构是Workflow（工作流）+ Prompt Chain（提示词链）。

我们将任务拆解为确定性的节点，在节点之间用代码控制流转，只把“需要模糊推理”的部分交给大模型。

import json
from typing import TypedDict, List

# 定义工作流的状态
class AgentState(TypedDict):
user_input: str
extracted_data: dict
analysis_result: str
final_output: dict
error: str

class FinancialAnalysisWorkflow:
def __init__(self, llm_client):
self.llm = llm_client

def run(self, user_input: str) –> dict:
state = AgentState(user_input=user_input, error="")

try:
# 节点1：意图识别与数据提取
state = self._extract_data(state)
if state.get("error"): return self._fallback(state)

# 节点2：大模型推理分析
state = self._analyze(state)
if state.get("error"): return self._fallback(state)

# 节点3：结构化输出校验
state = self._format_output(state)

return state["final_output"]

except Exception as e:
state["error"] = str(e)
return self._fallback(state)

def _extract_data(self, state: AgentState) –> AgentState:
# 使用Function Calling或JSON模式强制提取
prompt = f"从以下文本中提取财务数据，输出JSON：\\n{state['user_input']}"
response = self.llm.chat(prompt, response_format={"type": "json_object"})

try:
state["extracted_data"] = json.loads(response)
except json.JSONDecodeError:
state["error"] = "数据提取失败，格式非JSON"
return state

def _analyze(self, state: AgentState) –> AgentState:
prompt = f"基于以下数据生成分析报告：\\n{json.dumps(state['extracted_data'])}"
# 限制最大长度，防止模型无限发散
state["analysis_result"] = self.llm.chat(prompt, max_tokens=500)
return state

def _format_output(self, state: AgentState) –> AgentState:
# 强约束格式化，并在代码层面强行插入免责声明
state["final_output"] = {
"analysis": state["analysis_result"],
"disclaimer": "本报告由AI生成，仅供参考，不构成投资建议。"
}
return state

def _fallback(self, state: AgentState) –> dict:
return {"analysis": "系统繁忙或数据异常，请稍后再试。", "error": state.get("error")}

工程经验：能用代码写死的（如拼装JSON、加免责声明、判断是否为空），绝对不要交给大模型做。 大模型只负责“语言理解和生成”，这才是它能保持稳定输出的关键。

三、 开发阶段：Prompt工程与防御性编程

在开发阶段，工程师的核心工作从“写业务逻辑”变成了“写Prompt”和“写校验规则”。

1. Prompt版本化

千万别把Prompt写在Java/Python的代码字符串里。一旦需求变动，你需要重新编译打包。把Prompt抽离到配置中心或数据库，像管理代码一样管理Prompt的版本。

class PromptManager:
def __init__(self, config_client):
self.config = config_client

def get_prompt(self, scene: str, version: str = "latest") –> str:
# 从配置中心拉取
return self.config.get(f"prompt_{scene}_{version}")

2. 防御性解析

永远不要相信大模型会100%返回合法的JSON。即使你用了json_object模式，也可能遇到截断、包含Markdown标记等问题。

import re

def safe_parse_json(text: str) –> dict:
"""防御性JSON解析"""
# 1. 去除Markdown代码块标记
text = re.sub(r'```json\\s*|\\s*```', '', text).strip()

# 2. 尝试直接解析
try:
return json.loads(text)
except json.JSONDecodeError:
pass

# 3. 尝试提取大括号内的内容
match = re.search(r'\\{.*\\}', text, re.DOTALL)
if match:
try:
return json.loads(match.group(0))
except json.JSONDecodeError:
pass

# 4. 彻底失败，返回默认值并记录日志
logger.error(f"JSON解析彻底失败: {text[:100]}")
return {"error": "PARSE_FAILED", "raw": text}

四、 测试阶段：从“断言”到“评估”

AI应用不能只写单元测试。因为同样的输入，模型可能给出不同的输出（即使Temperature设为0，底层算子也可能有微小的差异）。

AI原生的测试叫Evals（评估）。

class LLMJudge:
def __init__(self, evaluator_llm):
self.llm = evaluator_llm

def evaluate(self, query: str, expected: str, actual: str) –> dict:
prompt = f"""你是一个严格的裁判。请评估AI的回答质量。
用户问题：{query}
标准答案：{expected}
AI实际回答：{actual}

评估维度：
1. 事实准确性（是否出现幻觉）- 0到5分
2. 格式合规性（是否包含免责声明）- 0到5分
3. 语气与专业性 – 0到5分

请输出JSON评分及理由："""
response = self.llm.chat(prompt)
return json.loads(response)

五、 上线与运维：可观测性即生命

AI应用上线后，如果没有完善的监控，就像在黑夜中闭眼开车。

必须记录的指标

• 业务指标：QPS、成功率、平均耗时。
• AI专属指标：Token消耗、首字延迟（TTFT，流式输出）、Prompt长度。
• 质量指标：解析失败率、兜底触发率、用户点踩率。

链路追踪（Trace）

每一次Agent调用，必须记录完整的上下文：用户的原始输入 -> 最终发给模型的Prompt -> 模型返回的原始文本 -> 经过解析后的数据 -> 最终给用户的输出。

import logging
import json

logger = logging.getLogger("agent_trace")

def trace_agent_call(task_id, user_input, prompt, raw_response, parsed_output):
# 在实际生产中，这里应该写入ELK或ClickHouse
trace_data = {
"task_id": task_id,
"input": user_input,
"prompt_sent": prompt,
"raw_response": raw_response,
"parsed_output": parsed_output
}
logger.info(f"AGENT_TRACE: {json.dumps(trace_data, ensure_ascii=False)}")

这样，当业务方跑来说“昨天下午这个回答有问题”时，你才能立刻拿着task_id查出到底是用户输入太奇葩、Prompt没写清、还是模型抽风，从而快速定位问题。

写在最后

AI原生应用（Agent）的开发，本质上是在**“用确定性的工程代码”去包裹“概率性的AI能力”**。

从需求定义边界，到设计Workflow拆解任务，再到防御性编程、LLM-as-a-Judge评估，最后到全链路可观测性——这套端到端的流程，才是将Agent从“好玩的Demo”变成“可靠的生产力工具”的必经之路。

不要迷信“全自动智能体”，在生产环境里，克制和约束往往比自由和创造更能创造商业价值。如果你的团队正准备上线第一个Agent，建议先从一个边界清晰的Workflow开始，把错误兜底和日志链路做扎实，再逐步引入更复杂的规划能力。