我用 Python + AI Agent 搭了一套自动干活系统:从指令到执行,全程零手动
我用 Python + AI Agent 搭了一套自动干活系统:从指令到执行,全程零手动
适合想让 AI 不只是"聊天"、而是真正帮你干活的开发者。
本文展示一套完整的 AI Agent 自动化系统,从接收指令到执行任务到返回结果,全程零手动干预,附可运行代码。
背景:AI 聊天 ≠ AI 干活
大多数人用 AI 的方式是:打开对话框 → 打字问问题 → 看回答 → 自己去执行。
这不叫自动化,这叫"AI 替你查资料"。
真正的自动化是:你说一句话,AI 自己去操作文件、调用 API、搜索信息、生成内容、输出结果。你只需要验收。
我花了 3 个月时间,用 Python + AI Agent 搭了一套系统,目前每天能自动完成这些任务:
| 任务 | 以前耗时 | 现在耗时 | 做法 |
|---|---|---|---|
| 搜集写作素材 | 30 分钟 | 2 分钟 | AI 搜索知识库 + 整理 |
| 写初稿 | 2 小时 | 10 分钟 | AI 生成 + 知识库增强 |
| 生成封面图 | 15 分钟 | 0.5 秒 | Python PIL 脚本 |
| 排版 | 10 分钟 | 0 | 自动排版模板 |
| 合规检查 | 5 分钟 | 0 | AI 自动检查 |
| 推送到草稿箱 | 5 分钟 | 1 分钟 | API 自动推送 |
一篇文章从选题到进草稿箱,总耗时从 3 小时降到 15 分钟。
系统架构
整个系统分 4 层:
用户指令(一句话描述任务)
↓
AI Agent(理解意图、规划步骤)
↓
工具层(MCP 协议连接外部工具)
↓
执行层(Python 脚本执行具体操作)
↓
结果返回(输出到文件/文档/草稿箱)
核心思路:AI 负责"想",Python 负责"做"。
AI Agent 不直接操作文件系统,而是通过 MCP 协议调用 Python 脚本。这样既利用了 AI 的理解能力,又保证了操作的可靠性。
核心模块 1:指令解析器
用户输入一句话,AI 需要理解:要做什么、用什么工具、输出到哪里。
# 指令解析的 prompt 模板
TASK_PARSER_PROMPT = """
你是一个任务解析器。用户的指令是:{user_input}
请解析出以下信息,以 JSON 格式返回:
{
"task_type": "collect|write|publish|analyze",
"target": "任务目标描述",
"tools_needed": ["需要用到的工具列表"],
"output_format": "md|docx|html|json",
"output_path": "输出路径"
}
"""
示例:
用户输入:帮我搜集关于 MCP 协议的最新资料,整理成一篇技术文章
AI 解析结果:
{
"task_type": "write",
"target": "MCP 协议技术文章",
"tools_needed": ["knowledge_base_search", "web_search", "article_writer"],
"output_format": "md",
"output_path": "output/mcp-article.md"
}
核心模块 2:工具注册表
所有可用的工具在这里注册,AI Agent 根据任务选择合适的工具:
TOOL_REGISTRY = {
"knowledge_base_search": {
"description": "搜索知识库中的相关资料",
"function": search_knowledge_base,
"params": ["query", "limit"]
},
"web_search": {
"description": "搜索互联网获取最新信息",
"function": search_web,
"params": ["query", "num_results"]
},
"article_writer": {
"description": "根据素材生成技术文章",
"function": write_article,
"params": ["topic", "materials", "style"]
},
"cover_generator": {
"description": "用 PIL 生成封面图",
"function": generate_cover,
"params": ["title", "style", "output_path"]
},
"compliance_checker": {
"description": "检查文章是否符合平台规范",
"function": check_compliance,
"params": ["article_path", "platform"]
},
"publisher": {
"description": "将文章推送到平台草稿箱",
"function": publish_to_draft,
"params": ["article_path", "cover_path", "platform"]
}
}
核心模块 3:任务编排器
AI Agent 解析完指令后,编排器负责按顺序调用工具:
async def execute_task(parsed_task):
"""按步骤执行任务"""
results = {}
for step in parsed_task["steps"]:
tool_name = step["tool"]
tool = TOOL_REGISTRY[tool_name]
# 调用工具
print(f"执行: {step['description']}")
result = await tool["function"](**step["params"])
# 检查结果
if result.get("error"):
print(f"步骤失败: {result['error']}")
# 尝试重试或跳过
if step.get("optional"):
continue
else:
raise Exception(f"关键步骤失败: {tool_name}")
results[tool_name] = result
return results
核心模块 4:知识库检索
这是让 AI 写出"有你味道"文章的关键。没有知识库,AI 只能写泛泛而谈的内容。
def search_knowledge_base(query, limit=5):
"""从本地知识库搜索相关资料"""
# 1. 从知识库目录搜索匹配文件
results = []
kb_path = Path("knowledge_base/")
for file in kb_path.rglob("*.md"):
content = file.read_text(encoding="utf-8")
# 简单关键词匹配(生产环境用向量搜索)
score = sum(1 for word in query.split() if word in content)
if score > 0:
results.append({
"file": str(file),
"score": score,
"excerpt": content[:500]
})
# 2. 按相关度排序,取 top N
results.sort(key=lambda x: x["score"], reverse=True)
return results[:limit]
实战:一句话完成一篇文章
完整流程演示:
用户输入:帮我写一篇关于"Python 自动化办公"的技术文章,发到 CSDN 草稿箱
系统执行过程:
步骤 1/6:解析指令
→ task_type: write
→ tools: [knowledge_base, web_search, article_writer, cover_gen, compliance, publisher]
步骤 2/6:搜索知识库
→ 找到 3 篇相关资料
步骤 3/6:搜索互联网
→ 找到 5 篇最新相关文章
步骤 4/6:生成文章
→ 输出:output/python-automation.md(约 3000 字)
步骤 5/6:生成封面图 + 合规检查
→ 封面图:cover/python-automation.png
→ 合规检查:通过
步骤 6/6:推送到 CSDN 草稿箱
→ 成功,文章 ID: xxxxx
总耗时:约 2 分钟
多 Agent 协作:复杂任务拆分
当任务比较复杂时,单个 Agent 处理不过来。这时候需要多 Agent 协作:
# 多 Agent 编排示例
async def multi_agent_workflow(topic):
"""多 Agent 协作完成复杂任务"""
# Agent 1:选题研究员
researcher = Agent("researcher", tools=["web_search", "knowledge_base"])
research_result = await researcher.run(f"搜集关于 {topic} 的最新资料")
# Agent 2:文章写手
writer = Agent("writer", tools=["article_writer"])
article = await writer.run(f"基于以下资料写文章: {research_result}")
# Agent 3:质量审核员
reviewer = Agent("reviewer", tools=["compliance_checker"])
review_result = await reviewer.run(f"检查这篇文章: {article}")
if review_result["passed"]:
# Agent 4:发布员
publisher = Agent("publisher", tools=["cover_generator", "publisher"])
await publisher.run(f"生成封面并发布: {article}")
else:
# 打回修改
article = await writer.run(f"根据审核意见修改: {review_result['feedback']}")
return article
每个 Agent 有自己的工具集和职责,互不干扰。这也是我前面文章里讲过的"Agent 流水线"的实际代码。
踩坑记录
坑 1:AI 理解指令出错
症状:用户说"帮我写篇文章",AI 理解成"帮我搜索文章"。
原因:指令太模糊,AI 无法准确判断意图。
解决:在 prompt 里加 few-shot 示例,让 AI 看几个"指令→解析结果"的对照。
坑 2:工具调用顺序错误
症状:AI 先调用"发布"工具,再调用"生成文章"工具。
原因:编排器没有检查依赖关系。
解决:给每个工具加 depends_on 字段,编排器按依赖顺序执行。
坑 3:知识库搜索不准
症状:搜"Python 自动化",返回了一堆无关的文件。
原因:简单的关键词匹配太粗糙。
解决:用向量搜索替代关键词匹配。先用 Embedding 模型把知识库内容向量化,搜索时计算余弦相似度。
坑 4:AI Agent 死循环
症状:AI 调用工具失败后不断重试,陷入死循环。
原因:没有设置最大重试次数。
解决:每个工具设置 max_retries: 3,超过就报错退出。
坑 5:输出格式不稳定
症状:同样是写文章,有时输出 Markdown,有时输出纯文本。
原因:prompt 里没有明确指定输出格式。
解决:在 prompt 里明确写"输出必须是 Markdown 格式,第一行是 # 标题"。
坑 6:多 Agent 通信开销大
症状:4 个 Agent 协作,总耗时比单 Agent 还长。
原因:Agent 之间传递的上下文太大,每次都要把完整资料传过去。
解决:Agent 之间只传摘要,不传全文。资料存在共享存储里,Agent 通过 ID 引用。
完整项目结构
ai-workflow/
├── agents/ # Agent 定义
│ ├── researcher.py # 选题研究员
│ ├── writer.py # 文章写手
│ ├── reviewer.py # 质量审核员
│ └── publisher.py # 发布员
├── tools/ # 工具注册
│ ├── knowledge_base.py # 知识库搜索
│ ├── web_search.py # 网络搜索
│ ├── article_writer.py # 文章生成
│ ├── cover_gen.py # 封面生成
│ ├── compliance.py # 合规检查
│ └── publisher.py # 平台发布
├── knowledge_base/ # 本地知识库
│ ├── writing_style.md # 写作风格规范
│ ├── templates/ # 文章模板
│ └── references/ # 参考资料
├── config/
│ ├── tools.yaml # 工具配置
│ └── agents.yaml # Agent 配置
└── main.py # 主入口
总结
3 条核心经验:
-
AI 负责"想",脚本负责"做"。不要让 AI 直接操作文件系统,让它输出"要做什么"的指令,Python 脚本去执行。这样更可靠,也好调试。
-
知识库是灵魂。没有知识库的 AI Agent 只是一个聊天机器人,有了知识库它才能写出有你个人风格的内容。
-
从单 Agent 开始,逐步扩展到多 Agent。不要一上来就搞复杂的多 Agent 系统,先把单 Agent 的流程跑通,再拆分。
你有没有搭过类似的自动化系统?遇到过什么问题?评论区聊聊。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐
所有评论(0)