一人公司 AI 产品搭建实战：从0到1的极简主义生存法则

my_ai_product/
├── frontend/                 # Streamlit UI (简单到爆)
│   └── app.py
├── backend/                  # FastAPI 核心逻辑
│   ├── main.py
│   ├── rag_engine.py         # 知识库检索
│   ├── agent_workflow.py     # Agent 工作流
│   └── models.py
├── data/                     # 本地数据存储
│   ├── chroma_db/            # 向量库文件
│   └── uploads/              # 用户上传文件
├── prompts/                  # 提示词工程（单独放，方便调优）
│   └── listing_system.txt
├── requirements.txt
└── .env                      # 存API Key，打死不提交Git

# backend/rag_engine.py
import chromadb
from chromadb.utils import embedding_functions
import pandas as pd
from sentence_transformers import SentenceTransformer

class LocalRAG:
    def __init__(self):
        # 1. 初始化本地Embedding模型（小，快，且免费）
        self.embed_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
            model_name="paraphrase-multilingual-MiniLM-L12-v2"
        )
        # 2. 启动持久化客户端
        self.client = chromadb.PersistentClient(path="./data/chroma_db")
        
        # 3. 获取或创建Collection
        self.collection = self.client.get_or_create_collection(
            name="amazon_listings",
            embedding_function=self.embed_fn
        )
        
    def seed_database(self):
        """只需运行一次，灌入竞品数据"""
        df = pd.read_csv("data/seed.csv")
        for idx, row in df.iterrows():
            # 将标题+描述合并，作为检索文本
            text = f"{row['title']} {row['bullet_points']}"
            self.collection.add(
                documents=[text],
                metadatas=[{"rank": row['rank'], "category": row['cat']}],
                ids=[f"doc_{idx}"]
            )
        print(f"成功灌入 {len(df)} 条数据")

    def retrieve(self, query, top_k=5):
        """检索最相似的竞品写法"""
        results = self.collection.query(
            query_texts=[query],
            n_results=top_k
        )
        return results['documents'][0]  # 返回文本列表

# backend/agent_workflow.py
import os
from openai import OpenAI
from .rag_engine import LocalRAG

class ListingAgent:
    def __init__(self):
        self.client = OpenAI(
            api_key=os.getenv("DEEPSEEK_API_KEY"),
            base_url="https://api.deepseek.com/v1"  # 兼容OpenAI
        )
        self.rag = LocalRAG()
        
    def generate(self, product_info: dict):
        # 1. 先RAG检索同类优秀Listing作为参考
        context = self.rag.retrieve(product_info['category'] + " " + product_info['keywords'])
        
        # 2. 组装超级提示词
        system_prompt = f"""
        你是一名亚马逊高级运营专家。请参考以下竞品优秀文案风格：
        -----------
        {context}
        -----------
        
        请根据用户提供的产品信息，生成5个不同风格的标题（各不超过200字符）和5点描述。
        必须符合亚马逊A9算法偏好（埋词、短句、情感触发）。
        直接输出JSON格式，不要有多余废话。
        """

        user_content = f"""
        产品名称：{product_info['name']}
        核心卖点：{product_info['features']}
        目标人群：{product_info['target_audience']}
        价格区间：{product_info['price_range']}
        """

        # 3. 调用LLM（强制性JSON输出）
        response = self.client.chat.completions.create(
            model="deepseek-chat",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_content}
            ],
            response_format={"type": "json_object"},  # DeepSeek支持
            temperature=0.8
        )
        
        return response.choices[0].message.content

# backend/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from .agent_workflow import ListingAgent
import json

app = FastAPI()
agent = ListingAgent()

class ProductRequest(BaseModel):
    name: str
    features: str
    target_audience: str = ""
    price_range: str = ""
    category: str = "Electronics"  # 默认

@app.post("/api/generate_listing")
async def generate(request: ProductRequest):
    try:
        result_json = agent.generate(request.dict())
        # 直接透传JSON给前端
        return json.loads(result_json)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/health")
async def health():
    return {"status": "OK", "message": "一人公司AI正在运行"}

# frontend/app.py
import streamlit as st
import requests
import json

# 页面配置
st.set_page_config(page_title="AI跨境Listing神器", layout="centered")
st.title("🚀 一键生成爆款Listing")
st.caption("基于百万级Best Seller数据训练，专为一人卖家打造")

# 侧边栏：用户信息与计费
with st.sidebar:
    st.header("⚡️ 账户余额")
    # 这里接入Lemon Squeezy的购买校验逻辑
    if st.button("💰 充值10美金 (生成20次)"):
        # 跳转到支付链接
        st.markdown("[点击支付](https://your-store.lemonsqueezy.com/checkout)")
    
    st.divider()
    st.caption("剩余生成次数: 15")

# 主表单
with st.form("product_form"):
    col1, col2 = st.columns(2)
    with col1:
        name = st.text_input("产品名称", placeholder="例如：无线降噪耳机")
        category = st.selectbox("品类", ["Electronics", "Home", "Clothing", "Beauty"])
    with col2:
        price = st.text_input("价格区间", placeholder="$25 - $40")
        audience = st.text_input("目标人群", placeholder="通勤上班族")
    
    features = st.text_area("核心卖点（用逗号分隔）", placeholder="长续航, 主动降噪, 舒适佩戴")
    
    submitted = st.form_submit_button("✨ 立即生成5个版本", type="primary")

if submitted:
    if not name or not features:
        st.warning("请至少填写产品名称和卖点")
        st.stop()
    
    with st.spinner("正在分析竞品数据并生成...（约10秒）"):
        # 调用后端API
        response = requests.post(
            "http://localhost:8000/api/generate_listing",
            json={
                "name": name,
                "features": features,
                "target_audience": audience,
                "price_range": price,
                "category": category
            }
        )
        
        if response.status_code == 200:
            data = response.json()
            
            # 用标签页展示5个版本
            tabs = st.tabs([f"版本 {i+1}" for i in range(5)])
            for i, tab in enumerate(tabs):
                with tab:
                    st.subheader(f"📌 标题 Version {i+1}")
                    st.code(data['titles'][i], language="markdown")
                    
                    st.subheader("📝 五点描述")
                    for point in data['bullet_points'][i]:
                        st.write(f"- {point}")
                    
                    # 复制按钮
                    st.button(f"📋 复制此版本", key=f"copy_{i}")
        else:
            st.error("生成失败，请检查API配置或余额")

# backend/credit_manager.py
import redis
import os

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def check_and_deduct(user_id: str):
    """检查并扣除积分"""
    key = f"user:{user_id}:credits"
    credits = r.get(key)
    if credits is None:
        # 新用户送3次试用
        r.setex(key, 3600 * 24 * 7, 3)  # 7天过期
        return 3
    
    if int(credits) <= 0:
        return 0
    
    # 扣减一次
    r.decr(key)
    return int(credits) - 1

# backend/webhook.py
@app.post("/webhook/lemon")
async def lemon_webhook(request: Request):
    payload = await request.json()
    event = payload.get('meta', {}).get('event_name')
    
    if event == 'order_created':
        user_email = payload['data']['attributes']['user_email']
        # 给用户充值 20 次
        r.setex(f"user:{user_email}:credits", 3600*24*30, 20)  # 30天有效
        print(f"用户 {user_email} 充值成功")
    
    return {"ok": True}

# 1. 安装fc组件
npm install @alicloud/fc2 -g

# 2. 部署（一条命令）
fun deploy -y

# utils/weekly_report.py
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是一个创业教练，帮创始人写本周复盘。"},
        {"role": "user", "content": f"本周新增用户{new_users}，付费转化率{conversion}%，主要bug是{bug_list}。请写一份鼓舞人心的周报。"}
    ]
)
print(response.choices[0].message.content)