4 AI Automation Pipelines for a Solo Builder: Content, Replies, Data & Monitoring

The Pain: Content to write, users to reply to, data to watch, services to babysit. Each task is easy on its own; together they turn "building a startup" into "working for yourself."
What You'll Learn: The complete architecture of 4 AI pipelines — content generation, user auto-reply, data aggregation, and anomaly monitoring — with architecture diagrams, runnable code, docker-compose configs, and deployment guides you can reproduce on your own server.

---

If you're running an AI product solo, you'll hit this problem sooner or later: too many things to do, and only one of you.

Content needs writing, users need replies, data needs watching, services need babysitting. Every single one of these is easy in isolation. Add them all up, though, and it stops being "starting a business" and becomes "working for yourself."

This article isn't an operations retrospective — the project just started, there isn't much data yet. It's a technical architecture design. Picking up where the tech-stack article left off, we break down 4 AI automation pipelines that one person can run. Each pipeline has a complete architecture diagram and runnable code you can reproduce on your own server.

4 AI automation pipelines: Content, Replies, Data, Monitoring

The Big Picture: An MCP-Style Architecture One Person Can Run

All four pipelines share the same AI service layer and data layer — no duplication:

┌─────────────────────────────────────────────────────────────┐
│                     User Touch Layer                         │
│   WeChat Official Account ←→ Mini Program ←→ Blog ←→        │
│   WeChat Group Bot                                           │
└──────────────────────────────┬──────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────┐
│               Orchestration Layer (n8n + Celery Beat)        │
│  ┌───────────────┐ ┌───────────────┐ ┌───────────────┐      │
│  │ Content       │ │ Reply         │ │ Data          │      │
│  │ Pipeline      │ │ Pipeline      │ │ Pipeline      │      │
│  │ scheduled     │ │ message-      │ │ scheduled     │      │
│  │ trigger       │ │ triggered     │ │ aggregation   │      │
│  │ AI gen+publish│ │ AI reply      │ │ push to board │      │
│  └───────────────┘ └───────────────┘ └───────────────┘      │
│  ┌───────────────┐                                            │
│  │ Monitoring    │                                            │
│  │ Pipeline      │                                            │
│  │ health check  │                                            │
│  │ auto alert    │                                            │
│  └───────────────┘                                            │
└──────────────────────────────┬──────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────┐
│                  AI Service Layer (self-hosted + API)        │
│  ┌──────────────────────┐ ┌──────────────────────┐          │
│  │ Hermes Agent         │ │ Vector KB (Chroma)   │          │
│  │ Loop Engineering     │ │ persistent memory+RAG│          │
│  └──────────────────────┘ └──────────────────────┘          │
│  ┌──────────────────────┐ ┌──────────────────────┐          │
│  │ LLM API (DeepSeek)   │ │ Tool functions       │          │
│  │                      │ │ (search / compute)   │          │
│  └──────────────────────┘ └──────────────────────┘          │
└──────────────────────────────┬──────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────┐
│            Infrastructure Layer (one lightweight server)     │
│  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────────┐           │
│  │ n8n  │ │Redis │ │MySQL │ │Chroma│ │  Docker  │           │
│  └──────┘ └──────┘ └──────┘ └──────┘ └──────────┘           │
└─────────────────────────────────────────────────────────────┘

Core design principle: every pipeline runs in dual mode — event-driven + scheduled polling. Event-driven handles real-time needs; scheduled polling handles batch tasks. Both modes share the same AI service and data layer, with zero redundant construction.

Tech Stack Choices

Component	Choice	Why
Orchestration	n8n + Celery Beat	n8n for event-triggered visual workflows; Celery Beat for scheduled batch jobs
AI service	Hermes Agent + DeepSeek API	Hermes provides the Loop Engineering and tool-calling framework; DeepSeek is the LLM backend
Vector DB	Chroma	lightweight, embedded in the Python process, no separate deployment
Message queue	Redis	natively supported by both n8n and Celery; the lightest single-node option
Database	SQLite / MySQL	choose as needed; SQLite is enough in the early days
Containerization	Docker + docker-compose	unified environment, single-node friendly
Notifications	WeCom (WeChat Work) bot	free, supports Markdown messages

This whole stack runs on a 2-core / 4GB lightweight server (~$11/month). No Kubernetes, no microservices — one person can afford to operate it.

Let's break down each pipeline.

---

Pipeline 1: Content Production Automation

Goal: use AI to assist creation and cut the time spent writing from scratch.

Architecture

Content topic management
        ↓
Scheduled trigger (n8n / Celery Beat)
        ↓
AI draft generation (Hermes Agent + LLM)
        ↓  auto tool calls
        ├─ web search to verify facts
        ├─ code snippet generation
        └─ format conversion (Markdown → target platform)
        ↓
Store into publish queue
        ↓
Notify the creator for review

Code

The core is a scheduled workflow in the orchestrator. Content generation logic lives in one Python service:

# content_pipeline.py
# Deployment reference: /opt/opc/services/content_pipeline/
import json
import requests
from datetime import datetime
from hermes_agent import Agent, Loop, Tool

class ContentPipeline:
    """AI content production pipeline"""
    def __init__(self, system_prompt: str):
        self.agent = Agent(model="deepseek-chat", system_prompt=system_prompt)
        self.agent.register_tools([
            Tool(name="web_search", fn=self.search_web),
            Tool(name="run_code", fn=self.exec_python),
        ])

    def generate_article(self, topic: str, outline: str) -> dict:
        """Three steps: draft → polish → final review"""
        # Step 1: Draft
        draft = self.agent.run(
            f"Write a complete article on the topic '{topic}' with the outline '{outline}'. "
            f"Call web_search to verify any claims that need verification. "
            f"Call run_code to test whether any code snippets actually run."
        )
        # Step 2: Polish (Loop mode, iterative)
        loop = Loop(max_iterations=3)
        polished = loop.run(
            self.agent,
            [
                "Check the article for logical holes",
                "Optimize the title and opening appeal",
                "Make sure code examples are copy-paste runnable",
            ],
            context=draft,
        )
        # Step 3: Quality score
        quality = self.agent.run(
            f"Score the following article from 1 to 10. "
            f"If it's below 7, return revision suggestions: {polished}"
        )
        return {
            "title": self.extract_title(polished),
            "content": polished,
            "quality_score": self.parse_score(quality),
            "generated_at": datetime.now().isoformat(),
        }

    def search_web(self, query: str) -> str:
        r = requests.get("https://api.duckduckgo.com/", params={"q": query, "format": "json"})
        return r.text[:2000]

    def exec_python(self, code: str) -> str:
        import subprocess
        result = subprocess.run(["python3", "-c", code], capture_output=True, text=True, timeout=10)
        return result.stdout or result.stderr

Deployment

# Trigger with a systemd timer or n8n webhook; run once daily at 08:00
0 8 * * * cd /opt/opc/services/content_pipeline && python3 run_daily.py

Key principle: AI writes the draft, a human does the final review. The pipeline only generates drafts, stores them in the drafts box, and notifies the creator for review. A human must look at it before publishing — that step is not automatable.

Full Deployment: systemd Service + Timer

Besides cron, you can also manage the content pipeline's lifecycle with systemd:

# /etc/systemd/system/content-pipeline.service
[Unit]
Description=Content Pipeline Service
After=network.target docker.service

[Service]
Type=oneshot
User=opc
ExecStart=/usr/bin/python3 /opt/opc/services/content_pipeline/run_daily.py
WorkingDirectory=/opt/opc/services/content_pipeline
EnvironmentFile=/opt/opc/.env
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

# /etc/systemd/system/content-pipeline.timer
[Unit]
Description=Daily content generation at 8am

[Timer]
OnCalendar=*-*-* 08:00:00
Persistent=true

[Install]
WantedBy=timers.target

# Enable
sudo systemctl daemon-reload
sudo systemctl enable content-pipeline.timer
sudo systemctl start content-pipeline.timer

# Manual trigger for testing
sudo systemctl start content-pipeline.service
journalctl -u content-pipeline.service --no-pager   # view logs

n8n Workflow Config (optional)

If you use n8n for visual orchestration, the workflow nodes look like this:

1. Schedule Trigger (daily 08:00) ↓
2. HTTP Request → Python service (triggers content_pipeline.run_daily) ↓
3. Wait (wait for AI generation, up to 5 minutes) ↓
4. HTTP Request → query generation result ↓
5. If quality score ≥ 7 → WeChat Draft API → store in drafts → WeCom notification: "Today's draft is ready" Else → WeCom notification: "Today's draft didn't pass quality — pick a new topic"

n8n's advantage is visual configuration — no glue code needed. But if you'd rather skip n8n, a systemd timer + bash script works just as well.

---

Pipeline 2: User Interaction Automation

This pipeline handles automatic replies to user messages across WeChat Official Account, mini program, personal site, and other channels.

Architecture

User sends message → channel server pushes
        ↓
Webhook received (FastAPI)
        ↓
Vector KB query (RAG)
        ↓
AI intent understanding + reply generation
        ↓
Safety review filter
        ↓
Auto reply / flag for human

Core Code

# auto_reply.py
# FastAPI middleware service
from fastapi import FastAPI, Request
from chromadb import ChromaClient
from hermes_agent import Agent

app = FastAPI()
agent = Agent(model="deepseek-chat")
vector_db = ChromaClient().get_collection("knowledge_base")

# Intent classification
INTENT_CLASSIFIER = """
Classify the user message into one of the following labels, return only the label:
- product_inquiry (product inquiry)
- tech_support (technical support)
- cooperation (partnership intent)
- feedback (feedback / suggestions)
- unknown (cannot classify)

Message: {message}
"""

@app.post("/webhook/reply")
async def auto_reply(request: Request):
    data = await request.json()
    user_msg = data.get("Content", "")
    from_user = data.get("FromUserName", "")

    # Intent recognition
    intent = agent.run(INTENT_CLASSIFIER.format(message=user_msg)).strip()

    # Query knowledge base
    similar_docs = vector_db.query(query_texts=[user_msg], n_results=3)

    # Scenarios that need a human
    if intent == "unknown" or self._needs_human(intent, user_msg):
        notify_me(f"Needs human reply: {from_user}")
        return {"Content": "Got it, I'll get back to you shortly."}

    # Generate reply
    reply = agent.run(
        f"User asks: {user_msg}\nIntent: {intent}\n"
        f"Relevant knowledge: {similar_docs}\n"
        f"Reply in a friendly tone, keep it under 200 characters."
    )

    # Safety review
    if self._safety_check(reply):
        return {"Content": reply}
    else:
        notify_me(f"Reply blocked: {reply}")
        return {"Content": "Your question has been recorded."}

def _needs_human(intent: str, msg: str) -> bool:
    """Decide whether a human is needed"""
    if intent == "cooperation":
        return True
    if any(kw in msg for kw in ["refund", "complaint", "legal"]):
        return True
    return False

Deployment

# Run as a FastAPI service
uvicorn auto_reply:app --host 0.0.0.0 --port 8200

# Nginx reverse proxy
# location /webhook/ {
#     proxy_pass http://127.0.0.1:8200;
# }

The core value of this pipeline isn't "replacing humans" — it's filtering out standardized inquiries so humans only handle messages that truly need judgment and empathy.

WeChat Official Account Configuration

If you're connecting a WeChat Official Account, configure the server address in the account backend:

• Account backend → Settings & Development → Server config
• URL: https://your-domain.com/webhook/reply
• Token: custom (must match the token in code)
• Message encryption: plaintext mode (debugging) → secure mode (production)

WeChat server config verification code:

# wechat_verify.py
# Used to verify the server URL on first configuration
from flask import Flask, request, make_response
import hashlib

app = Flask(__name__)
WECHAT_TOKEN = "your_token_here"

@app.route("/webhook/reply", methods=["GET"])
def verify():
    """WeChat server verification endpoint"""
    signature = request.args.get("signature", "")
    timestamp = request.args.get("timestamp", "")
    nonce = request.args.get("nonce", "")
    echostr = request.args.get("echostr", "")

    # Sort lexicographically, then SHA1
    tmp_list = sorted([WECHAT_TOKEN, timestamp, nonce])
    tmp_str = "".join(tmp_list)
    tmp_str = hashlib.sha1(tmp_str.encode()).hexdigest()

    if tmp_str == signature:
        return echostr  # verification passed
    return "Verification failed", 403

@app.route("/webhook/reply", methods=["POST"])
def handle_message():
    """Receive user messages"""
    xml_data = request.data
    # Parse XML → extract Content/FromUserName
    # → call the auto_reply logic
    # → return the reply in XML format
    return reply_xml

Initializing the Chroma Knowledge Base

Auto-reply quality depends on knowledge base quality. Here's how to initialize it:

# 1. Install Chroma
pip install chromadb

# 2. Create the knowledge base collection
python3 << 'EOF'
import chromadb
client = chromadb.Client()
collection = client.create_collection(name="knowledge_base")

# Add documents: FAQ, product docs, technical docs, etc.
documents = [
    "Product A costs $99/month, or $999/year",
    "API rate limit is 100 calls/minute; contact the admin to raise it",
    "Server maintenance window is Wednesday 2:00-4:00 AM",
    # ... more documents
]
collection.add(
    documents=documents,
    ids=[f"doc_{i}" for i in range(len(documents))]
)
print(f"Knowledge base initialized, {len(documents)} documents")
EOF

# embedding_config.py
# Configure the embedding model and retrieval parameters
import chromadb
from chromadb.utils import embedding_functions

# Use sentence-transformers for local embeddings
sentence_ef = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="all-MiniLM-L6-v2"  # lightweight, ~1.5GB memory
)
client = chromadb.Client()
collection = client.get_or_create_collection(
    name="knowledge_base",
    embedding_function=sentence_ef,
)

def search_knowledge(query: str, top_k: int = 3) -> list:
    """Retrieve from the knowledge base"""
    results = collection.query(query_texts=[query], n_results=top_k)
    return results['documents'][0] if results['documents'] else []

Full Docker Compose Orchestration

All services are managed by docker-compose:

version: '3.8'
services:
  # Content pipeline
  content_pipeline:
    build: ./services/content_pipeline
    ports:
      - "8100:8100"
    env_file: .env
    volumes:
      - ./services/content_pipeline:/app
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "python3", "-c", "import requests; exit(0 if requests.get('http://localhost:8100/health').status_code == 200 else 1)"]
      interval: 60s
      timeout: 5s
      retries: 3

  # Auto-reply service
  auto_reply:
    build: ./services/auto_reply
    ports:
      - "8200:8200"
    env_file: .env
    environment:
      - CHROMA_PERSIST_DIR=/data/chroma
    volumes:
      - ./chroma_data:/data/chroma
      - ./services/auto_reply:/app
    depends_on:
      - redis
    restart: unless-stopped

  # Data collector
  data_collector:
    build: ./services/data_collector
    env_file: .env
    volumes:
      - ./data:/opt/opc/data
    restart: unless-stopped

  # Dashboard
  dashboard:
    image: nginx:alpine
    ports:
      - "9999:80"
    volumes:
      - ./dashboard:/usr/share/nginx/html:ro
    restart: unless-stopped

  # Infrastructure
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    restart: unless-stopped

  chroma:
    image: chromadb/chroma:latest
    ports:
      - "8000:8000"
    volumes:
      - chroma_data:/chroma/chroma
    environment:
      - IS_PERSISTENT=TRUE
    restart: unless-stopped

volumes:
  redis_data:
  chroma_data:

# Start everything with one command
docker-compose up -d
docker-compose ps   # confirm all services are healthy

---

Pipeline 3: Data Aggregation Automation

This pipeline solves the "I don't know how well I'm actually running" problem — collect key metrics on a schedule and push them to a dashboard.

Architecture

Celery Beat scheduled trigger
        ↓
Data collection
        ├─ server metrics (CPU / memory / disk)
        ├─ business data (DB aggregation)
        └─ third-party APIs (if any)
        ↓
Clean & aggregate
        ↓
Push to dashboard + time-series storage

Key Code

# data_collector.py
# Celery scheduled task
from celery import Celery
from datetime import datetime
import requests
import sqlite3

app = Celery("data_collector", broker="redis://localhost:6379/0")

@app.task
def collect_hourly():
    """Collect once per hour"""
    metrics = {}

    # 1. Server metrics
    try:
        cpu = requests.get("http://localhost:9100/metrics", timeout=5)
        metrics["cpu_usage"] = parse_cpu_metric(cpu.text)
        metrics["mem_usage"] = parse_mem_metric(cpu.text)
        metrics["disk_usage"] = parse_disk_metric(cpu.text)
    except:
        notify_alert("Failed to collect server metrics!")

    # 2. Business metrics
    db = sqlite3.connect("/opt/opc/data/metrics.db")
    cursor = db.cursor()
    cursor.execute("""
        SELECT COUNT(*) FROM api_logs
        WHERE created_at > datetime('now', '-1 hour')
    """)
    metrics["requests_past_hour"] = cursor.fetchone()[0]

    # 3. Write to the time-series store (ORM-style pseudocode:
    #    replaces the raw INSERT into hourly_metrics with a model object)
    db.add(HourlyMetric(
        collected_at=datetime.now().isoformat(),
        cpu=metrics.get("cpu_usage", 0),
        mem=metrics.get("mem_usage", 0),
        disk=metrics.get("disk_usage", 0),
        requests=metrics["requests_past_hour"],
    ))
    db.commit()
    db.close()

    # 4. Push to dashboard
    push_to_dashboard(metrics)

    # 5. Anomaly detection
    if metrics.get("cpu_usage", 0) > 85:
        notify_alert(f"CPU usage {metrics['cpu_usage']}%")
    if metrics.get("disk_usage", 0) > 90:
        notify_alert(f"Disk usage {metrics['disk_usage']}%")

def push_to_dashboard(metrics):
    requests.post(
        "http://localhost:9999/api/metrics",
        json={"timestamp": datetime.now().isoformat(), **metrics},
    )

def notify_alert(msg: str):
    """WeCom bot notification"""
    requests.post(
        "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx",
        json={"msgtype": "markdown", "markdown": {"content": msg}},
    )

Dashboard Frontend

Pair the collector with a simple real-time dashboard:

<!-- /opt/opc/dashboard/index.html -->
<!DOCTYPE html>
<html>
<head>
  <title>OPC Ops Dashboard</title>
  <style>
    body { font-family: -apple-system, sans-serif; background: #0f172a; color: #e2e8f0; padding: 24px; }
    .grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 16px; }
    .card { background: #1e293b; border-radius: 12px; padding: 20px; }
    .value { font-size: 2rem; font-weight: bold; color: #3b82f6; }
    .label { font-size: 0.85rem; color: #94a3b8; margin-top: 4px; }
    .ok { color: #22c55e; }
    .warn { color: #eab308; }
    .err { color: #ef4444; }
  </style>
</head>
<body>
  <h1>📊 OPC Ops Dashboard</h1>
  <div class="grid" id="metrics">
    <div class="card"><div class="value" id="cpu">-</div><div class="label">CPU usage</div></div>
    <div class="card"><div class="value" id="mem">-</div><div class="label">Memory usage</div></div>
    <div class="card"><div class="value" id="requests">-</div><div class="label">Requests (past hour)</div></div>
  </div>
  <script>
    setInterval(async () => {
      const res = await fetch('/api/metrics/latest');
      const data = await res.json();
      document.getElementById('cpu').textContent = (data.cpu ?? 0).toFixed(1) + '%';
      document.getElementById('mem').textContent = (data.mem ?? 0).toFixed(1) + '%';
      document.getElementById('requests').textContent = data.requests ?? 0;
    }, 30000);
  </script>
</body>
</html>

systemd Timer Config

Besides Celery, a lightweight systemd timer can also handle scheduled collection:

# /etc/systemd/system/data-collector.service
[Unit]
Description=Hourly data collector
After=network.target

[Service]
Type=oneshot
ExecStart=/usr/bin/python3 /opt/opc/services/data_collector/run.py
WorkingDirectory=/opt/opc/services/data_collector

[Install]
WantedBy=multi-user.target

# /etc/systemd/system/data-collector.timer
[Unit]
Description=Run data collector every hour

[Timer]
OnCalendar=hourly
Persistent=true

[Install]
WantedBy=timers.target

# Enable the timer
sudo systemctl enable data-collector.timer
sudo systemctl start data-collector.timer

---

Pipeline 4: Anomaly Monitoring Automation

This one is the lightest — but the most important, because it watches the other three.

Implementation

# docker-compose health check config
services:
  content_pipeline:
    build: ./services/content_pipeline
    healthcheck:
      test: ["CMD", "python3", "-c", "import requests; r = requests.get('http://localhost:8100/health'); exit(0 if r.status_code == 200 else 1)"]
      interval: 60s
      timeout: 5s
      retries: 3

  auto_reply:
    build: ./services/auto_reply
    healthcheck:
      test: ["CMD", "python3", "-c", "from app import check_health; exit(0 if check_health() else 1)"]
      interval: 120s
      timeout: 5s
      retries: 2

  data_collector:
    build: ./services/data_collector
    healthcheck:
      test: ["CMD-SHELL", "pgrep -f celery || exit 1"]
      interval: 300s
      timeout: 5s
      retries: 2

Pair it with Healthchecks.io for scheduled-task monitoring — the last step of each pipeline pings the API once; if no ping arrives within the timeout, an alert fires:

HEALTHCHECKS_URL = "https://hc-ping.com/your-uuid-here"

def ping_healthcheck(pipeline_name: str):
    """Ping once after each pipeline completes"""
    # No ping within 30 minutes → alert fires
    requests.get(f"{HEALTHCHECKS_URL}/{pipeline_name}")

Simple status dashboard (HTML):

<div class="pipeline-status">
  <div class="card content-pipeline">📝 Content pipeline</div>
  <div class="card reply-pipeline">💬 Reply pipeline</div>
  <div class="card data-pipeline">📊 Data pipeline</div>
  <div class="card monitor-pipeline">🔔 Monitor pipeline</div>
</div>

Four status cards. All green is the best morning there is.

Alert Notification Routing

The last step of anomaly monitoring is notification. Different severities need different channels:

Level	Example	Notification	Response time
P0 (emergency)	service completely down	phone + SMS + WeCom	within 15 min
P1 (severe)	API error rate > 5%	WeCom @all	within 30 min
P2 (warning)	CPU > 85%	WeCom message	within 2 hours
P3 (notice)	disk > 80%	dashboard yellow flag	next day

# alert_routing.py
from enum import Enum

class Severity(Enum):
    P0 = "p0"  # emergency
    P1 = "p1"  # severe
    P2 = "p2"  # warning
    P3 = "p3"  # notice

def route_alert(title: str, message: str, severity: Severity):
    """Choose the notification channel by severity"""
    payload = {
        "title": f"[{severity.value.upper()}] {title}",
        "message": message,
        "timestamp": datetime.now().isoformat(),
    }

    if severity == Severity.P0:
        # Notify on all channels simultaneously
        send_sms(payload)        # SMS
        make_phone_call(payload) # voice call
        send_wecom(payload)      # WeCom
    elif severity == Severity.P1:
        # WeCom @all
        payload["mentioned_list"] = ["@all"]
        send_wecom(payload)
    elif severity == Severity.P2:
        # Regular WeCom message
        send_wecom(payload)
    else:
        # Dashboard marker only
        write_to_dashboard(payload)

Self-Healing Attempts

For known anomaly types, you can configure automatic recovery scripts:

# self_heal.py
# Run by a systemd timer every 5 minutes

def try_heal(service_name: str, check_cmd: str, heal_cmd: str) -> bool:
    """If the service is unhealthy, attempt to recover it automatically"""
    import subprocess

    # Check
    check = subprocess.run(check_cmd, shell=True, capture_output=True)
    if check.returncode == 0:
        return True  # service is healthy

    # Attempt recovery
    print(f"⚠️ {service_name} unhealthy, attempting recovery...")
    heal = subprocess.run(heal_cmd, shell=True, capture_output=True)
    if heal.returncode == 0:
        notify_alert(f"✅ {service_name} auto-recovered")
        return True
    else:
        notify_alert(f"❌ {service_name} auto-recovery failed, manual intervention needed")
        return False

---

Summary

The core of these four pipelines isn't AI — it's engineering thinking:

• The content pipeline solves "the friction of starting from zero" — AI writes the draft, you do the second pass
• The reply pipeline solves "attention fragmentation" — standardized inquiries get filtered automatically
• The data pipeline solves "working blind, not knowing good from bad" — numeric feedback drives improvement
• The monitoring pipeline solves "the service died and nobody knew" — the unattended safety net

This setup is business-agnostic. Whether you're building an AI product, a SaaS tool, or doing content creation, pick the pipeline that hurts the most right now and start there. You don't need to go all-in at once — start with one, get it running, then add the next.

Best Practices for Starting From Zero

If this is your first time building this system, roll it out in this order:

1. Monitoring pipeline first (1 hour) — the lightest, with the biggest payoff. Just docker-compose + healthchecks + WeCom notifications, and you'll know the instant a service dies.
2. Data pipeline next (half a day) — collect server metrics and business data, push to a dashboard. With data, you know what to optimize.
3. Reply pipeline after that (1 day) — start with the simplest FAQ auto-reply, then grow the knowledge base.
4. Content pipeline last (continuous optimization) — content generation has the highest ROI but also needs the most tuning. Wait until the other three are stable.

Pitfalls to Avoid

• Never expose raw AI replies to users — always keep a safety review layer. Better to not reply than to reply wrong.
• The knowledge base is never "done" — spend 15 minutes a week reviewing questions users asked that AI couldn't answer, and add them.
• More monitoring isn't better — before adding every alert, ask "what will I do when I receive it?" An alert that produces no action is noise.
• AI-generated content ≠ AI-published content — AI writes the draft, you do the second pass. That baseline can't go.

The whole codebase is on GitHub and reproducible: git clone https://github.com/your-org/opc-automation. No need to write from scratch — tweak the config and it runs. Questions welcome in the comments.

Next article: From 30 Files to 2 — How I Slimmed My Agent System Down 90%. I'll break down the slimming framework step by step: what to do at each stage, and how.

About the author: Wu Ji (无记) — AI & digitalization practitioner focused on Agent engineering, Loop Engineering, and digital transformation. Practical, hands-on tutorials — follow along and it just works.