独立开发 AI 工作流:先把状态机跑通
独立产品不需要堆满功能,先把用户实际要完成的那一步磨顺。这篇只讨论一个问题:独立开发 AI 工作流:先把状态机跑通。
写作边界:围绕“独立开发 AI 工作流:先把状态机跑通”出现的数字、事故场景和性能结果均用于演示分析方法,不是特定项目的实测结论。落地时请记录版本、输入、资源、统计窗口和失败路径,再用自己的测试数据复核。
示例场景:1. 让 Agent 自动写项目的问题示例:30 个 Tool Calls 后直接死锁卡死
想依靠 AI 一键搞定项目开发,往往会卡在 Tool Calling 无限循环与上下文丢失上。
检查 Agent 执行日志,定位当时崩溃的具体现场:
cat /var/log/agent-runner/task-exec.log | grep -E "\\[ToolCall\\]|\\[Error\\]|\\[Retry\\]" | tail -n 25
命令行反馈的异常信息暴露了工程缺陷:
[ToolCall] FetchWebpage url="https://example.com/article"
[ToolCall] ParseHTML content_length=45210
[Error] JSONSchemaValidationFailed: expected 'summary' field in output
[Retry] Triggering Agent auto-repair prompt (Attempt 1)
[ToolCall] FetchWebpage url="https://example.com/article"
[Retry] Triggering Agent auto-repair prompt (Attempt 2)
…
[Fatal] Maximum Tool Execution Limit (30) Exceeded. Stack Overflow.
根因非常清晰:Agent 调用的解析工具返回格式轻微偏离了预期 Schema,而系统缺乏确定的数据校验与拦截机制,只是盲目依赖 Agent 自我修复 Prompt。结果 Agent 不断重复发起网络请求,最终触发死锁和 Token 消耗熔断。
示例场景:2. 最小可运行架构(MVA)设计:拆解系统边界与状态机
避免大模型瞎干的最好方式,是工程师亲自掌控系统架构与状态变更逻辑。
针对“网页抓取与结构化笔记”这个真实任务,我们需要将架构精简为四个职责单一的实体模块,而不是交由单一的庞大 Agent 自行发挥:
组件职责被严格切分:
示例场景:3. 任务拆解与工具调用的防死锁流程
把任务拆细后,还需要在 Agent 工具调用层加上硬性的防死锁与限速防线。
传统的 Tool Calling 是在 Prompt 里写一段说明让 LLM 自由选择工具。但在生产工程中,应限制:**每个状态
如果 Tool 执行失败,应返回结构化错误并进入人工处理或降级分支,不允许模型沿用错误参数继续重试。
示例场景:4. 可落地的轻量 Agent 工作流与组件隔离代码
下面是用 Python 实现的最小可运行架构调度器,结合了 JSON Schema 强制校验、Tool 执行限流以及显式状态机:
import json
import time
from typing import Dict, Any, Optional
from pydantic import BaseModel, Field, ValidationError
# 1. 定义数据传输契约 Schema
class NoteResultSchema(BaseModel):
title: str = Field(description="文章标题")
summary: str = Field(description="核心要点总结")
tags: list[str] = Field(description="关键词标签")
class TaskContext(BaseModel):
url: str
state: str = "INIT"
raw_html: Optional[str] = None
clean_text: Optional[str] = None
result: Optional[NoteResultSchema] = None
tool_retry_count: int = 0
class MVATaskExecutor:
MAX_RETRY = 2
def __init__(self, llm_client, fetch_tool):
self.llm_client = llm_client
self.fetch_tool = fetch_tool
def run_task(self, target_url: str) -> Dict[str, Any]:
ctx = TaskContext(url=target_url)
# 状态 1: 确定性 Fetch
ctx = self._step_fetch(ctx)
# 状态 2: 确定性 Parse
ctx = self._step_parse(ctx)
# 状态 3: 限制性 LLM 总结
ctx = self._step_summarize(ctx)
return ctx.result.model_dump()
def _step_fetch(self, ctx: TaskContext) -> TaskContext:
ctx.state = "FETCHING"
try:
# 确定性工具调用,不给 LLM 介入机会
ctx.raw_html = self.fetch_tool.get(ctx.url)
ctx.state = "FETCHED"
return ctx
except Exception as e:
raise RuntimeError(f"[MVA Error] Fetch failed: {str(e)}")
def _step_parse(self, ctx: TaskContext) -> TaskContext:
ctx.state = "PARSING"
# 纯 Python 文本提纯,剔除 HTML 脚本与样式
ctx.clean_text = self.fetch_tool.extract_text(ctx.raw_html)
ctx.state = "PARSED"
return ctx
def _step_summarize(self, ctx: TaskContext) -> TaskContext:
ctx.state = "SUMMARIZING"
prompt = f"请总结以下文本,输出严格符合 JSON 格式:\\n{ctx.clean_text[:3000]}"
while ctx.tool_retry_count < self.MAX_RETRY:
try:
# LLM 仅在此处调用,且开启 response_format 约束
response_str = self.llm_client.generate(prompt, response_format={"type": "json_object"})
parsed_json = json.loads(response_str)
# 强校验返回的数据结构
validated_result = NoteResultSchema(**parsed_json)
ctx.result = validated_result
ctx.state = "COMPLETED"
return ctx
except (json.JSONDecodeError, ValidationError) as err:
ctx.tool_retry_count += 1
print(f"[MVA Warning] Retrying LLM output validation (Attempt {ctx.tool_retry_count}): {err}")
time.sleep(0.5)
raise RuntimeError("[MVA Fatal] LLM failed to produce valid structure after retries.")
通过这套架构,独立开发者能够牢牢把控核心状态节点。代码中没有复杂的递归与自发工具匹配,系统,运行成本直线上升降到了原来的 10%。
示例场景:5. 独立开发者的架构止损规则:不要把架构主导权交给 Prompt
独立完成一个项目产品化的过程中,开发者需要建立以下三条架构防护防线:
- 从具体任务切入,拒绝宏大叙事:不要一开始就想做“通用 Agent”,先跑通“抓取单网页生成笔记”或“解析单一格式 PDF”的具体流程。
- 确定性逻辑归代码,非确定性语言归大模型:文本解析、网络请求、数据库读写全部用传统代码写死,大模型只用来做语言翻译、摘要提炼和意图识别。
- 限制单次 Task 的最大 Tool Calls 深度:在系统层设定硬性限制,单个似切断死循环消耗 Token 的后路。
搞清楚代码和模型的边界,才是独立开发者快速上线高质量产品的核心密码。





