从零搭建 OPC 一人公司：Python + AI Agent 全栈自动化实战（附完整代码）

摘要：2026 年，OPC（One Person Company，一人公司）已从概念走向规模化落地，全国 20+ 城市出台专项支持政策，市场规模预计突破 8000 亿元。但大多数 OPC 创业者仍停留在"手动调 AI"的阶段——问一句答一句，效率天花板很低。本文将以一个真实的 OPC 创业场景为例，手把手教你用 Python + AI Agent 搭建一套可自动运行的"数字员工团队"，覆盖内容生产、线索管理、智能客服、数据复盘四大核心模块，附全部源码。

一、OPC 一人公司的本质：不是"一个人干活"，是"一个人指挥一个系统"

OPC 的核心定义由江苏苏州在 2025 年 11 月的"人工智能 OPC 大会"上首次提出，其核心逻辑是：

人做决策 / 创意 / 战略，AI 做执行 / 标准化 / 流程化。

但 90% 的 OPC 创业者还停留在这样的使用模式：

plaintext
1
2
打开 ChatGPT → 提问 → 复制回答 → 粘贴到文档 → 结束

这不是 OPC，这叫"一个人用高级计算器"。

真正的 OPC 应该是一个自动化闭环系统：

plaintext
1
2
输入（选题/需求/线索） → AI 自动处理 → 人工审核关键节点 → 自动输出（内容/回复/报告） → 数据反馈 → 优化迭代

本文要做的，就是用 Python 把这个闭环写出来、跑起来、自动化运行。

二、OPC 技术架构全景图

在动手写代码之前，先看清整个系统的骨架：

plaintext
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
┌─────────────────────────────────────────────────────────────┐
│ OPC 一人公司技术架构 │
├─────────────┬──────────────┬──────────────┬─────────────────┤
│ 内容生产引擎 │ 线索管理系统 │ 智能客服Agent │ 数据复盘面板 │
│ │ │ │ │
│ • 选题分析 │ • 多渠道收集 │ • 知识库检索 │ • 营收统计 │
│ • 文案生成 │ • 自动分级 │ • 多轮对话 │ • 内容效果 │
│ • 自动排版 │ • 跟进提醒 │ • 人工转接 │ • 客户转化 │
│ • 定时发布 │ • 状态追踪 │ • 记录沉淀 │ • 优化建议 │
├─────────────┴──────────────┴──────────────┴─────────────────┤
│ 核心调度层（Python + LLM） │
│ LLM 封装 · 工具调用 · 记忆管理 · 任务编排 │
├─────────────────────────────────────────────────────────────┤
│ 基础设施层 │
│ SQLite 存储 · 定时任务 · API 接口 · 文件管理 │
└─────────────────────────────────────────────────────────────┘

技术选型一览

表格
层级 技术方案 选型理由
LLM 大脑 DeepSeek API（兼容 OpenAI 接口） 国内最便宜的可用方案，¥1/百万 token
编程语言 Python 3.11+ 生态最丰富，AI 库最全
数据存储 SQLite + JSON 文件 零运维，单人足够用
定时调度 APScheduler 轻量级，支持 cron 表达式
知识库 本地 Markdown + 向量检索 不依赖外部服务
Web 框架 FastAPI 提供 API 接口，方便扩展

三、项目初始化与环境搭建

3.1 项目结构

bash
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
opc_company/
├── config/
│ └── settings.py # 全局配置
├── core/
│ ├── llm.py # LLM 封装层
│ ├── agent.py # Agent 核心调度
│ └── memory.py # 记忆管理
├── modules/
│ ├── content_engine.py # 内容生产引擎
│ ├── lead_manager.py # 线索管理系统
│ ├── customer_bot.py # 智能客服
│ └── analytics.py # 数据复盘
├── knowledge/ # 知识库目录
│ ├── faq/
│ ├── products/
│ └── cases/
├── outputs/ # 自动产出目录
│ ├── articles/
│ ├── reports/
│ └── templates/
├── data/
│ └── opc.db # SQLite 数据库
├── main.py # 启动入口
└── requirements.txt

3.2 安装依赖

bash
1
2
3
4
5
6
7
8
9

创建虚拟环境

python -m venv opc-env
source opc-env/bin/activate # Linux/Mac

opc-env\Scripts\activate # Windows

安装核心依赖

pip install openai python-dotenv apscheduler fastapi uvicorn
sqlalchemy aiosqlite httpx jieba

3.3 配置文件

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

config/settings.py

import os
from dotenv import load_dotenv

load_dotenv()

LLM 配置（以 DeepSeek 为例，兼容 OpenAI 接口）

LLM_CONFIG = {
“api_key”: os.getenv(“DEEPSEEK_API_KEY”, “sk-xxx”),
“base_url”: os.getenv(“LLM_BASE_URL”, “https://api.deepseek.com/v1”),
“model”: os.getenv(“LLM_MODEL”, “deepseek-chat”),
“temperature”: 0.7,
“max_tokens”: 4096,
}

数据库

DB_PATH = “data/opc.db”

知识库路径

KNOWLEDGE_DIR = “knowledge”
OUTPUT_DIR = “outputs”

内容发布配置

CONTENT_PLATFORMS = [“csdn”, “juejin”, “zhihu”]

定时任务配置

SCHEDULE_CONFIG = {
“content_generate”: “0 9 * * 1-5”, # 工作日 9:00 生成内容
“lead_review”: “0 10 * * *”, # 每天 10:00 审查线索
“daily_report”: “0 20 * * *”, # 每天 20:00 生成日报
}

四、核心模块一：LLM 封装层

所有上层模块都依赖一个稳定、可复用的 LLM 调用层。

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

core/llm.py

import httpx
import json
from typing import List, Optional
from config.settings import LLM_CONFIG

class LLMClient:
“”“统一的 LLM 调用封装，支持多轮对话和工具调用”“”

def __init__(self, config: dict = None): self.config = config or LLM_CONFIG self.client = httpx.Client(timeout=60.0) self.conversation_history: List[dict] = [] def chat(self, prompt: str, system_prompt: str = "", context: str = "") -> str: """ 单轮对话，适合自动化场景 """ messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) if context: messages.append({ "role": "system", "content": f"以下是参考信息，请基于此回答：\n{context}" }) messages.append({"role": "user", "content": prompt}) return self._call_api(messages) def chat_with_memory(self, prompt: str, system_prompt: str = "") -> str:

五、核心模块二：内容生产引擎

这是 OPC 最核心的自动化模块——让 AI 自动完成"选题→写文→排版→存档"的全流程。

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

modules/content_engine.py

import os
import json
from datetime import datetime
from core.llm import LLMClient

class ContentEngine:
“”“OPC 内容自动化生产引擎”“”

def __init__(self, llm: LLMClient, output_dir: str = "outputs/articles"): self.llm = llm self.output_dir = output_dir os.makedirs(output_dir, exist_ok=True) def generate_topics(self, niche: str, count: int = 5) -> list: """ 第一步：基于垂直领域生成选题 """ schema = { "topics": [ { "title": "文章标题（技术博客风格，带数字和关键词）", "angle": "切入角度（50字以内）", "target_audience": "目标读者", "estimated_reads": "预估阅读量级：高/中/低" } ] } prompt = f"""你是一个 CSDN 技术博客的资深选题策划。

当前时间：{datetime.now().strftime(‘%Y年%m月’)}
垂直领域：{niche}

请生成 {count} 个高点击率的技术博客选题。要求：

1. 标题必须有数字、有关键词、有痛点

六、核心模块三：线索管理系统

OPC 不是写完文章就结束了——你需要把读者变成客户。

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222

modules/lead_manager.py

import sqlite3
import json
from datetime import datetime, timedelta
from typing import List, Optional
from core.llm import LLMClient

class LeadManager:
“”“OPC 线索管理与自动跟进系统”“”

def __init__(self, llm: LLMClient, db_path: str = "data/opc.db"): self.llm = llm self.db_path = db_path self._init_db() def _init_db(self): """初始化数据库""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS leads ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT NOT NULL, -- 来源平台：csdn/zhihu/juejin/wechat name TEXT, -- 联系人 company TEXT, -- 公司 industry TEXT, -- 行业 problem TEXT NOT NULL, -- 具体问题/需求 urgency TEXT DEFAULT 'medium', -- 紧急程度：high/medium/low status TEXT DEFAULT 'new', -- 状态：new/contacted/quoted/closed/lost solution_type TEXT, -- 可交付方案：consulting/agent/workflow/training notes TEXT, -- 跟进记录 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS lead_activities ( id INTEGER PRIMARY KEY AUTOINCREMENT, lead_id INTEGER NOT NULL, action TEXT NOT NULL, content TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (lead_id) REFERENCES leads(id) ) """) conn.commit() conn.close() def add_lead(self, source: str, problem: str, name: str = None, company: str = None, industry: str = None) -> int: """ 添加新线索，AI 自动分析紧急程度和方案类型 """ # 用 AI 分析线索质量 analysis = self.llm.structured_output( f"""分析以下客户线索，判断紧急程度和适合的交付方案。

来源平台：{source}
客户问题：{problem}
客户行业：{industry or ‘未知’}

请分析：

1. 紧急程度（high/medium/low）

2. 推荐交付方案类型

3. 建议的首次回复策略"“”,
   {
   “urgency”: “high/medium/low”,
   “solution_type”: “consulting/agent/workflow/training”,
   “reply_strategy”: “建议的首次回复要点（100字内）”
   }
   )

   conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(""" INSERT INTO leads (source, name, company, industry, problem, urgency, solution_type, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( source, name, company, industry, problem, analysis.get("urgency", "medium"), analysis.get("solution_type", "consulting"), analysis.get("reply_strategy", "") )) lead_id = cursor.lastrowid # 记录活动 cursor.execute(""" INSERT INTO lead_activities (lead_id, action, content) VALUES (?, 'created', ?) """, (lead_id, f"新线索录入 | AI分析: {json.dumps(analysis, ensure_ascii=False)}")) conn.commit() conn.close() print(f"✅ 线索 #{lead_id} 已录入 | 紧急度: {analysis.get('urgency')}") return lead_id

   def get_pending_leads(self, status: str = “new”) -> List[dict]:
   “”“获取待处理线索”“”
   conn = sqlite3.connect(self.db_path)
   conn.row_factory = sqlite3.Row
   cursor = conn.cursor()

   cursor.execute(""" SELECT * FROM leads WHERE status = ? ORDER BY CASE urgency WHEN 'high' THEN 1 WHEN 'medium' THEN 2 ELSE 3 END, created_at DESC """, (status,)) leads = [dict(row) for row in cursor.fetchall()] conn.close() return leads

   def generate_follow_up(self, lead_id: int) -> str:
   “”"
   AI 生成跟进话术
   “”"
   conn = sqlite3.connect(self.db_path)
   conn.row_factory = sqlite3.Row
   cursor = conn.cursor()

   cursor.execute("SELECT * FROM leads WHERE id = ?", (lead_id,)) lead = dict(cursor.fetchone()) conn.close() if not lead: return "❌ 线索不存在" prompt = f"""根据以下客户线索信息，生成一段专业且有温度的首次跟进话术。

客户信息：

• 来源：{lead[‘source’]}
• 问题：{lead[‘problem’]}
• 行业：{lead[‘industry’] or ‘未知’}
• 紧急程度：{lead[‘urgency’]}
• 推荐方案：{lead[‘solution_type’]}
• 已有备注：{lead[‘notes’] or ‘无’}

要求：

1. 开头点明你在哪里看到他的需求

2. 简要说明你能帮他解决什么问题

3. 给出一个具体的下一步行动建议

4. 语气专业但不生硬，像一个靠谱的技术合伙人

5. 控制在 200 字以内"“”

   reply = self.llm.chat(prompt, system_prompt="你是一个 OPC 一人公司的创始人，擅长 AI 自动化解决方案。") # 记录活动 conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(""" INSERT INTO lead_activities (lead_id, action, content) VALUES (?, 'follow_up_generated', ?) """, (lead_id, reply)) conn.commit() conn.close() return reply

   def daily_lead_report(self) -> str:
   “”"
   生成每日线索报告
   “”"
   conn = sqlite3.connect(self.db_path)
   cursor = conn.cursor()

   # 统计数据 cursor.execute("SELECT COUNT(*) FROM leads") total = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM leads WHERE status = 'new'") new_leads = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM leads WHERE status = 'closed'") closed = cursor.fetchone()[0] cursor.execute(""" SELECT urgency, COUNT(*) FROM leads WHERE status IN ('new', 'contacted') GROUP BY urgency """) urgency_dist = dict(cursor.fetchall()) cursor.execute(""" SELECT source, COUNT(*) FROM leads GROUP BY source ORDER BY COUNT(*) DESC """) source_dist = dict(cursor.fetchall()) conn.close() report_data = { "total_leads": total, "new_today": new_leads, "closed": closed, "urgency_distribution": urgency_dist, "source_distribution": source_dist, } report = self.llm.chat( f"""基于以下 OPC 线索数据，生成一份简洁的每日运营报告。

数据：
{json.dumps(report_data, ensure_ascii=False, indent=2)}

报告需包含：

1. 数据概览（3行以内）

2. 重点关注项（哪些线索需要优先处理）

3. 趋势判断和建议

4. 明日行动计划"“”,
   system_prompt=“你是一个 OPC 创业者的运营助手，报告要简洁实用，不要废话。”
   )

   return report

七、核心模块四：智能客服 Agent

用知识库 + LLM 搭建一个能回答客户常见问题的 AI 客服。

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

modules/customer_bot.py

import os
import json
from datetime import datetime
from core.llm import LLMClient

class CustomerBot:
“”“OPC 智能客服 Agent - 基于知识库的问答系统”“”

def __init__(self, llm: LLMClient, knowledge_dir: str = "knowledge"): self.llm = llm self.knowledge_dir = knowledge_dir self.knowledge_base = {} self._load_knowledge() def _load_knowledge(self): """ 加载知识库（Markdown 文件） 目录结构： knowledge/ ├── faq/ # 常见问题 ├── products/ # 产品说明 └── cases/ # 案例资料 """ for category in os.listdir(self.knowledge_dir): category_path = os.path.join(self.knowledge_dir, category) if os.path.isdir(category_path): self.knowledge_base[category] = {} for file in os.listdir(category_path): if file.endswith(".md"): filepath = os.path.join(category_path, file) with open(filepath, "r", encoding="utf-8") as f: self.knowledge_base[category][file] = f.read() total_docs = sum(len(docs) for docs in self.knowledge_base.values()) print(f"📚 知识库加载完成：{total_docs} 篇文档") def search_knowledge(self, query: str, top_k: int = 3) -> str: """ 简单的关键词检索（生产环境可升级为向量检索） """ import jieba query_words = set(jieba.cut(query)) scores = [] for category, docs in self.knowledge_base.items(): for filename, content in docs.items(): # 简单的关键词匹配打分 content_words = set(jieba.cut(content)) overlap = len(query_words & content_words) score = overlap / max(len(query_words), 1) if score > 0.1: scores.append((score, category, filename, content[:1000])) # 按分数排序，取 top_k scores.sort(reverse=True) if not scores: return "未找到相关知识库内容" results = [] for score, cat, fname, excerpt in scores[:top_k]: results.append(f"### [{cat}/{fname}] (相关度: {score:.2f})\n{excerpt}...") return "\n\n---\n\n".join(results) def answer(self, question: str, conversation_id: str = None) -> dict: """ 回答客户问题：先检索知识库，再让 LLM 基于知识库回答 """ print(f"\n🤖 收到客户问题：{question}") # Step 1: 检索知识库 context = self.search_knowledge(question) # Step 2: 判断是否能回答 if context == "未找到相关知识库内容": return { "answer": "感谢您的提问！这个问题我需要让我们的技术专家来为您解答。" "请留下您的联系方式，我会在 2 小时内回复您。", "confidence": "low", "need_human": True, "source": None } # Step 3: 基于知识库生成回答 system_prompt = """你是 OPC 一人公司的智能客服助手。

回答规则：

1. 只基于提供的知识库内容回答，不要编造

2. 如果知识库信息不足以完整回答，明确告知并建议联系人工

3. 涉及价格、合同、效果承诺时，必须提示"建议与我们的顾问直接沟通确认"

4. 语气专业友好，像一个靠谱的技术顾问

5. 回答简洁，控制在 300 字以内"“”

   prompt = f"""客户问题：{question}

以下是从知识库中检索到的相关信息：
{context}

请基于以上信息回答客户的问题。如果信息不足以回答，请说明。“”"

answer = self.llm.chat(prompt, system_prompt) return { "answer": answer, "confidence": "high" if len(answer) > 100 else "medium", "need_human": "联系" in answer or "顾问" in answer, "source": context[:200] if context else None } def add_knowledge(self, category: str, filename: str, content: str): """添加知识库文档""" dir_path = os.path.join(self.knowledge_dir, category) os.makedirs(dir_path, exist_ok=True) filepath = os.path.join(dir_path, filename) with open(filepath, "w", encoding="utf-8") as f: f.write(content) # 重新加载 self.knowledge_base.setdefault(category, {})[filename] = content print(f"📚 知识库已更新：{category}/{filename}")

独立运行测试

ifname== “main”:
llm = LLMClient()
bot = CustomerBot(llm)

# 模拟客户提问 result = bot.answer("你们的 AI 自动化方案能帮我省多少人力成本？") print(f"\n回答：{result['answer']}") print(f"置信度：{result['confidence']}") print(f"需人工：{result['need_human']}")

八、核心模块五：数据复盘面板

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

modules/analytics.py

import sqlite3
import json
from datetime import datetime, timedelta
from core.llm import LLMClient

class AnalyticsDashboard:
“”“OPC 数据复盘面板”“”

def __init__(self, llm: LLMClient, db_path: str = "data/opc.db"): self.llm = llm self.db_path = db_path def get_weekly_stats(self) -> dict: """获取本周核心数据""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 本周新增线索 cursor.execute(""" SELECT COUNT(*) FROM leads WHERE created_at >= date('now', '-7 days') """) new_leads_week = cursor.fetchone()[0] # 本周关闭线索 cursor.execute(""" SELECT COUNT(*) FROM leads WHERE status = 'closed' AND updated_at >= date('now', '-7 days') """) closed_week = cursor.fetchone()[0] # 转化率 cursor.execute("SELECT COUNT(*) FROM leads") total = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM leads WHERE status = 'closed'") closed_total = cursor.fetchone()[0] conversion_rate = (closed_total / total * 100) if total > 0 else 0 # 来源分布 cursor.execute(""" SELECT source, COUNT(*) as cnt FROM leads GROUP BY source ORDER BY cnt DESC """) sources = {row[0]: row[1] for row in cursor.fetchall()} # 内容产出统计 article_dir = "outputs/articles" article_count = 0 if os.path.exists(article_dir): article_count = len([f for f in os.listdir(article_dir) if f.endswith('.md')]) conn.close() return { "period": "近 7 天", "new_leads": new_leads_week, "closed_leads": closed_week, "total_leads": total, "conversion_rate": f"{conversion_rate:.1f}%", "sources": sources, "articles_published": article_count, } def generate_weekly_report(self) -> str: """ AI 生成周报 """ stats = self.get_weekly_stats() report = self.llm.chat( f"""基于以下 OPC 一人公司本周运营数据，生成一份简洁的周报。

数据：
{json.dumps(stats, ensure_ascii=False, indent=2)}

周报要求：

1. 核心指标概览（不超过 5 行）
2. 本周亮点和问题
3. 渠道效率分析（哪个来源 ROI 最高）
4. 下周重点任务建议（不超过 3 条）
5. 用数据说话，不要空话

格式：Markdown"“”,
system_prompt=“你是一个 OPC 创业者的运营分析师，风格简洁务实。”
)

return report

import os

ifname== “main”:
llm = LLMClient()
dashboard = AnalyticsDashboard(llm)

stats = dashboard.get_weekly_stats() print("📊 本周数据：") print(json.dumps(stats, ensure_ascii=False, indent=2)) report = dashboard.generate_weekly_report() print("\n📝 本周周报：") print(report)

九、总调度：把所有模块串起来

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

main.py

“”"
OPC 一人公司 - 自动化运营系统启动入口
“”"
import json
import os
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger

from core.llm import LLMClient
from modules.content_engine import ContentEngine
from modules.lead_manager import LeadManager
from modules.customer_bot import CustomerBot
from modules.analytics import AnalyticsDashboard
from config.settings import SCHEDULE_CONFIG, LLM_CONFIG

初始化核心组件

llm = LLMClient()
content_engine = ContentEngine(llm)
lead_manager = LeadManager(llm)
customer_bot = CustomerBot(llm)
analytics = AnalyticsDashboard(llm)

def job_content_production():
“”“定时任务：内容生产”“”
print(f"\n⏰ [{datetime.now()}] 触发内容生产任务…“)
result = content_engine.auto_produce(“OPC 一人公司 + AI Agent 自动化”)
print(f"📄 产出结果：{result.get(‘status’)}”)

def job_lead_review():
“”“定时任务：线索审查”“”
print(f"\n⏰ [{datetime.now()}] 触发线索审查任务…“)
pending = lead_manager.get_pending_leads()
print(f"📋 待处理线索：{len(pending)} 条”)

for lead in pending[:5]: # 每次最多处理 5 条 reply = lead_manager.generate_follow_up(lead['id']) print(f" → 线索 #{lead['id']}: {lead['problem'][:30]}...") print(f" 建议回复：{reply[:80]}...")

def job_daily_report():
“”“定时任务：日报生成”“”
print(f"\n⏰ [{datetime.now()}] 触发生成日报任务…")
report = analytics.generate_weekly_report()

# 保存报告 os.makedirs("outputs/reports", exist_ok=True) date_str = datetime.now().strftime("%Y%m%d") filepath = f"outputs/reports/daily_{date_str}.md" with open(filepath, "w", encoding="utf-8") as f: f.write(report) print(f"📊 日报已保存：{filepath}")

def setup_scheduler():
“”“配置定时任务”“”
scheduler = BackgroundScheduler()

# 内容生产：工作日 9:00 scheduler.add_job( job_content_production, CronTrigger.from_crontab(SCHEDULE_CONFIG["content_generate"]) ) # 线索审查：每天 10:00 scheduler.add_job( job_lead_review, CronTrigger.from_crontab(SCHEDULE_CONFIG["lead_review"]) ) # 日报：每天 20:00 scheduler.add_job( job_daily_report, CronTrigger.from_crontab(SCHEDULE_CONFIG["daily_report"]) ) scheduler.start() print("✅ 定时任务调度器已启动") return scheduler

def main():
print(“=” * 60)
print(“🚀 OPC 一人公司 - 自动化运营系统”)
print(f" 启动时间：{datetime.now().strftime(‘%Y-%m-%d %H:%M:%S’)}“)
print(”=" * 60)

# 启动定时任务 scheduler = setup_scheduler() print("\n📌 系统运行中...") print(" - 内容生产：工作日 09:00") print(" - 线索审查：每天 10:00") print(" - 运营日报：每天 20:00") print("\n按 Ctrl+C 退出\n") try: # 保持主线程运行 import time while True: time.sleep(60) except KeyboardInterrupt: print("\n👋 系统关闭") scheduler.shutdown()

ifname== “main”:
main()

十、运行效果展示

10.1 启动系统

bash
1
2
3
4
5
6
7
8
9
10
11
12
13
14
$ python main.py

============================================================
🚀 OPC 一人公司 - 自动化运营系统
启动时间：2026-07-03 10:00:00

📚 知识库加载完成：12 篇文档
✅ 定时任务调度器已启动

📌 系统运行中…

• 内容生产：工作日 09:00
• 线索审查：每天 10:00
• 运营日报：每天 20:00

10.2 自动产出文章

plaintext
1
2
3
4
5
6
7
8
9
10
11
12
🚀 开始自动生产内容 | 领域：OPC 一人公司 + AI Agent 自动化

📋 Step 1: 生成选题…
✅ 选中选题：用 Python + AI Agent 搭建一人公司全栈自动化系统（附源码）

✍️ Step 2: 撰写文章（约需 2-3 分钟）…
💾 Step 3: 格式化并保存…
✅ 文章已保存：outputs/articles/20260703_0900_用_Python_AI_Agent_搭建一.md

🎉 内容生产完成！

10.3 线索自动分析

plaintext
1
2
3
4
5
6
7
⏰ [2026-07-03 10:00:00] 触发线索审查任务…
📋 待处理线索：3 条
→ 线索 #5: 想用 AI 自动回复客户咨询…
建议回复：您好！在 CSDN 上看到您对 AI 客服自动化的需求…
→ 线索 #6: 需要搭建社区团购小程序…
建议回复：您好！了解到您在做社区团购业务…

十一、成本估算：一个人运营的月度开销

表格
项目 方案 月费用
LLM API DeepSeek（日均 50 次调用） ¥30-80
服务器 轻量云服务器 2C4G ¥50-100
域名 .com 域名 ¥6/月（年付 ¥68）
数据库 SQLite（本地，零成本） ¥0
知识库存储 本地文件（可迁 OSS） ¥0-10
总计 ¥86-196/月

对比传统雇人方案：

表格
岗位 月薪（含社保）
内容运营 ¥6,000-10,000
客服专员 ¥4,000-6,000
数据分析师 ¥8,000-12,000
总计 ¥18,000-28,000/月

OPC 模式下的运营成本仅为传统团队的 0.5%-1%。

十二、总结与展望

本文做了什么

搭建了 LLM 统一封装层：支持单轮、多轮、结构化 JSON 输出
实现了内容生产引擎：选题→写作→格式化→保存全自动
构建了线索管理系统：自动录入、AI 分析、话术生成
开发了智能客服 Agent：知识库检索 + LLM 回答
完成了数据复盘面板：自动统计 + AI 周报
串起了定时调度：所有模块自动化运行，无需人工干预

OPC 的核心心法

OPC 不是一个人干所有活，而是：

plaintext
1
2
你定义系统 → 系统自动执行 → 你审核关键节点 → 系统持续优化

技术只是手段，真正的杠杆是把可复用的流程沉淀为代码和模板。今天写的一个内容生产脚本，明天就能自动帮你产出 100 篇文章。

下一步演进方向

表格
方向 技术方案 优先级
向量知识库 ChromaDB / FAISS 替代关键词检索 ⭐⭐⭐
多 Agent 协作 CrewAI / AutoGen 构建专家团 ⭐⭐⭐
MCP 连接器 打通飞书/企微/邮件 ⭐⭐
前端面板 Next.js + FastAPI 构建管理后台 ⭐⭐
支付与合同 Stripe / 支付宝 + 电子签章 ⭐

参考资源

OPC 一人公司创业全景指南 - CSDN
用 Codex 搭建最小 OPC 系统
一人公司技术栈治理实践 - InfoQ
用 Claude Code 构建一人公司：九个自动化系统
OpenClaw 实战指南：一人公司架构师 - CSDN

作者注：本文所有代码均可直接复制运行（需替换 API Key）。完整项目代码已上传至 GitHub，欢迎 Star 和 Issue。OPC 不是终点，而是每个技术人走向"系统换钱"而非"时间换钱"的起点。