博客 / 詳情

返回

XHS Agent 開發博客:用 AI 自動化小紅書內容運營的技術實踐

項目介紹

XHS Agent 是一個面向小紅書內容創作者和運營者的 AI 自動化工具。它能根據你設定的運營目標,自動生成圖文內容、AI 配圖,並按照最佳發佈時間定時發佈到小紅書賬號,全程無需人工干預。

核心能力包括:基於 LLM 的文案生成、多模型 AI 圖片生成(支持海報和真實照片兩種風格)、參考圖片素材庫管理(上傳後由視覺模型自動識別標註)、運營總管 AI 制定 7 天發佈計劃、APScheduler 定時調度發佈、WxPusher 微信通知推送。整個系統通過 Web 管理界面操作,支持多賬號管理,Docker 一鍵部署。

項目地址:GitHub項目鏈接


項目背景

小紅書內容運營有一個典型的痛點:高頻、高質量、風格一致,三者很難同時滿足。人工運營要麼頻率跟不上,要麼風格飄忽,要麼素材複用率低。時間一長,賬號更新就會斷更或者內容質量參差不齊。

XHS Agent 的目標是把這條鏈路自動化。從"我想發什麼"到"筆記已發佈",中間所有環節——文案生成、配圖生成、定時發佈、消息通知——全部由 AI 和調度系統接管。用户只需要設定運營目標,剩下的交給系統。

整個項目的後端用 FastAPI + aiosqlite 構建,全程異步,調度用 APScheduler 的 AsyncIOScheduler 直接跑在同一個事件循環裏,不需要額外的 worker 進程。前端是 React + Vite + Ant Design,構建產物直接掛載到 FastAPI 的靜態目錄,一個容器、一個端口搞定部署。包管理用 uv,比 pip/poetry 快一個數量級,依賴解析幾乎是秒級的。


多級 AI 編排流水線

整個項目最核心的設計不是某一個 AI 調用,而是把多個 AI 能力串聯成一條有明確分工的流水線。一次內容生成要經過三個階段:文本生成、提示詞優化、圖片生成,每個階段的輸出是下一個階段的輸入。

第一階段,text_service​ 調用 LLM 生成筆記的標題、正文、話題標籤,同時做一個關鍵決策:這篇筆記的配圖用 poster​(海報設計)還是 photo(真實照片)風格。這個決策基於兩個維度——主題語義(探店/穿搭/旅行傾向 photo,知識分享/產品推薦傾向 poster)和參考圖標註(如果用户上傳的參考圖被視覺模型識別為插畫風格,則強制 poster)。

# 將參考圖標註注入 user_prompt,引導 LLM 做風格決策
if ref_annotations:
    user_prompt += "\n\n參考圖片素材標註(請據此判斷視覺風格):"
    for ann in ref_annotations:
        cat = ann.get("category", "style")
        text = ann.get("annotation", "")
        if text:
            user_prompt += f"\n  - [{cat}] {text}"

第二階段,prompt_agent 拿到風格決策後,從預設的 8 個模板(4 種 photo + 4 種 poster)中篩選出對應風格的 4 個,再讓 LLM 為每張圖選擇最合適的模板並填充場景細節。這一層的存在是為了解決"初步提示詞質量不穩定"的問題——直接讓文本 LLM 寫圖片提示詞,質量參差不齊,而通過模板約束加上專門的提示詞工程師角色,輸出質量會穩定很多。

第三階段,image_service 拿到優化後的提示詞列表,根據風格決定併發還是串行生成圖片,最終返回圖片 URL 列表。這種分層設計的好處是每一層職責單一,可以獨立替換模型或策略,不會牽一髮動全身。


Poster 風格的串行生成策略

生成一組海報時,有一個很實際的問題:如果併發生成 3-6 張圖,每張圖的風格會有細微差異,整體看起來不協調,像是來自不同設計師的作品。

解決方案是 poster 模式下改為串行生成,並且把第一張圖的結果作為後續圖片的參考圖傳入,形成"風格錨定"效果。第一張圖確定了整體的色調、排版風格和視覺語言,後續每張圖在生成時都能看到第一張,自然會往同一個方向靠。

if is_poster:
    # 第一張:只用用户上傳的參考圖
    first_result = await _generate_single(built_prompts[0], size, ref_image_urls)
    results = [first_result]

    # 把第一張結果追加到參考圖列表,作為後續圖片的風格錨點
    anchor_urls = list(ref_image_urls or [])
    if first_result.url:
        anchor_urls.append(first_result.url)

    for prompt in built_prompts[1:]:
        result = await _generate_single(prompt, size, anchor_urls)
        results.append(result)
else:
    # photo 模式不需要風格一致性約束,直接併發
    tasks = [_generate_single(p, size, ref_image_urls) for p in built_prompts]
    results = await asyncio.gather(*tasks)

photo 模式不需要這個約束,真實照片本身的多樣性反而是優點,所以直接併發生成,速度也更快。


參考圖片系統與視覺識別

這個功能解決了"AI 不知道你的品牌長什麼樣"的問題。用户可以為每個賬號上傳參考圖片組,系統用 GLM-4.6V 視覺模型自動分析並生成文字標註,後續生圖時這些標註會作為上下文傳入,讓生成結果更貼近用户的視覺風格。

參考圖分為 5 種分類,每種分類對應不同的分析角度。風格參考組關注色調、光線、構圖和濾鏡風格;人物形象組關注外貌特徵、穿着和氣質;產品素材組關注產品外觀、包裝和品質感;場景環境組關注空間佈局和氛圍;品牌元素組關注 Logo、主色調和視覺規範。每種分類的 system prompt 都是專門定製的,引導視覺模型從對應角度輸出有用的描述。

# vision_service.py:按分類定製分析角度
CATEGORY_PROMPTS: dict[str, str] = {
    "style": (
        "你是一位專業的小紅書視覺分析師。請從【風格參考】角度整體描述這組圖片:\n"
        "1. 整體視覺調性(色調、光線、氛圍)\n"
        "2. 構圖方式和排版特點\n"
        "3. 濾鏡/後期風格傾向\n"
        "4. 適合的小紅書內容方向\n"
        "請用中文回答,控制在 300 字以內。"
    ),
    # person、product、scene、brand 各有對應的定製 prompt
    ...
}

上傳體驗上有一個細節:圖片上傳後立即返回,視覺識別在後台異步執行,不阻塞用户操作。實現上用 asyncio.create_task 觸發後台任務,前端輪詢狀態,識別完成後自動展示標註結果。

async def save_group(...) -> dict:
    # COS 上傳 + 數據庫寫入(立即完成)
    ...
    # 觸發後台識別,不等待
    asyncio.create_task(_run_group_vision(group_id, category, user_prompt))
    # 立即返回 pending 狀態
    return group_record

標註完成後,這些文字描述會在兩個地方發揮作用:一是傳給 text_service​ 影響風格決策,二是傳給 prompt_agent 融入圖片提示詞。與此同時,參考圖的 COS URL 也會直接傳給圖片生成 API 作為圖生圖的參考,讓視覺風格的傳遞更直接。


運營總管 AI

manager_service 是一個扮演"運營總監"角色的 AI,職責是分析賬號現狀、制定 7 天內容發佈計劃,併為每條排期從素材庫中選擇合適的參考圖片組。

它的 system prompt 裏注入了小紅書運營的核心規律:最佳發佈時間段(早高峯 7-9 點、午休 12-13:30、晚高峯 18-22 點)、內容節奏(乾貨/教程類 + 生活記錄類 + 種草類交替發佈)、發佈頻率建議(新賬號 1-2 篇/天,成熟賬號 1-3 篇/天)。這些規律不是讓 AI 自己摸索的,而是直接作為先驗知識注入,保證輸出的計劃符合平台規律。

在調用 AI 之前,系統會先併發拉取賬號的歷史數據——近 30 天的筆記統計和最近發佈的筆記列表。這兩個接口底層調用的是同步的 xhs SDK,用 asyncio.to_thread 包裝後併發執行,不阻塞事件循環。

async def fetch_account_stats(cookie: str, user_id: str = "") -> dict:
    stats, notes = await asyncio.gather(
        asyncio.to_thread(_get_stats, cookie),
        asyncio.to_thread(_get_recent_notes, cookie, user_id),
        return_exceptions=True,  # 任一失敗不影響另一個
    )
    return {
        "stats": stats if not isinstance(stats, Exception) else [],
        "recent_notes": notes if not isinstance(notes, Exception) else [],
    }

歷史數據彙總成摘要後傳給 AI,讓它知道哪類內容點贊/收藏更高,從而在新計劃裏優先複製爆款方向。參考圖選擇的邏輯也寫在 system prompt 裏:產品推廣內容必選產品素材組,需要保持人物一致性的內容必選人物形象組,每條內容最多選 3 個參考組。


APScheduler 與 asyncio 的集成

調度器選用 AsyncIOScheduler​,直接跑在 FastAPI 的 asyncio 事件循環裏,不需要額外的線程或進程。每個定時任務用 DateTrigger​ 指定精確的執行時間,觸發後調用 execute_scheduled_post 執行完整的內容生成和發佈鏈路。

服務重啓後內存中的調度任務全部丟失,這是一個容易被忽略的工程細節。解決方案是在 FastAPI 的 lifespan 啓動鈎子裏,從數據庫加載所有 pending 狀態的任務重新註冊到調度器。對於已經過期的任務,直接標記為失敗,不補發——避免重啓後批量觸發造成賬號異常。

async def reload_pending_jobs() -> None:
    posts = await list_scheduled_posts()
    now = datetime.now()
    for post in posts:
        if post["status"] != "pending":
            continue
        run_time = datetime.strptime(post["scheduled_at"], "%Y-%m-%d %H:%M")
        if run_time <= now:
            # 過期任務直接標記失敗,不補發
            async with get_db() as db:
                await db.execute(
                    "UPDATE scheduled_posts SET status='failed', error='服務重啓時任務已過期' WHERE id=?",
                    (post["id"],),
                )
                await db.commit()
            continue
        _add_job(post["id"], run_time)

添加任務時還有一個防重複的處理:每次添加前檢查 job_id 是否已存在,存在則先移除再添加。misfire_grace_time=300 允許任務在預定時間後 5 分鐘內補發,應對短暫的服務抖動而不至於直接丟任務。

def _add_job(post_id: int, run_time: datetime) -> None:
    job_id = f"post_{post_id}"
    if scheduler.get_job(job_id):
        scheduler.remove_job(job_id)
    scheduler.add_job(
        _run_post,
        trigger=DateTrigger(run_date=run_time),
        args=[post_id],
        id=job_id,
        misfire_grace_time=300,
    )

SQLite Schema 自動遷移

項目用 SQLite 存儲所有數據,迭代過程中表結構會頻繁變化。引入 Alembic 這類遷移工具對單機小項目來説太重了,所以實現了一個輕量的自動遷移機制:啓動時對比 schema 定義與實際表結構,缺失的字段自動用 ALTER TABLE ADD COLUMN 補充。

async def init_db():
    async with get_db() as db:
        # 建表,IF NOT EXISTS 保證冪等
        await db.executescript(CREATE_TABLES_SQL)
        # 檢查並補充缺失字段
        await _migrate_columns(db)

這個方案只支持新增字段,不支持字段改名或刪除,但對於這個項目的迭代節奏來説已經夠用。每次加新功能需要新字段時,只需要在 schema 定義里加上,重啓服務自動生效,不需要手動執行任何遷移腳本。


前後端一體化部署

前端 React 應用構建後直接放到後端的 static/​ 目錄,FastAPI 掛載靜態文件目錄,並對根路由兜底返回 index.html,支持前端的客户端路由。

app.mount("/static", StaticFiles(directory="static"), name="static")

@app.get("/")
async def index():
    return FileResponse("static/index.html")

所有後端接口以 /api​ 為前綴,前端直接請求同域接口,不存在跨域問題,也不需要 nginx 做反向代理。Docker 部署只需要一個容器、一個端口,docker-compose up -d​ 啓動後直接訪問 http://localhost:8000


踩過的幾個坑

圖片生成 API 的耗時遠超普通接口,單張圖片通常需要 30-120 秒。httpx 的默認超時是 5 秒,不顯式設置的話請求會直接報錯。這個問題很隱蔽,因為在本地測試時網絡條件好、模型響應快,往往不會觸發,上線後才暴露。解決方式是對圖片生成的 client 單獨設置 180 秒超時。

async with httpx.AsyncClient(timeout=180.0) as client:
    response = await client.post(...)

LLM 的輸出格式穩定性也是一個持續的問題。即使在請求裏指定了 response_format: {"type": "json_object"},部分模型偶爾還是會在 JSON 外面包一層 markdown 代碼塊,或者輸出字段名與約定不完全一致。處理方式是在解析前做字符串清洗,同時對每個字段都做兜底處理,避免 KeyError 導致整個流程崩潰。

圖片下載也有類似的穩定性問題。生成完成後需要把圖片 URL 下載到本地臨時文件再上傳到小紅書,這個下載步驟偶爾會因為網絡抖動失敗。加了指數退避重試,最多重試 3 次,間隔分別是 1 秒、2 秒、4 秒,覆蓋大多數短暫的網絡問題。

async def download_image_to_tmp(url: str, max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            ...
        except Exception:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)

開發經歷

第一天:從一個能跑的原型開始

項目的第一個提交是 2026 年 2 月 19 日中午,距離現在不到一週。初始版本非常簡單:FastAPI 提供一個生成接口,調用 LLM 生成文案,調用圖片 API 生成配圖,然後用 xhs SDK 發佈。前端是一個手寫的靜態 HTML,塞在 static/index.html 裏,沒有構建工具,沒有組件化,就是能用。

這個階段的目標只有一個:跑通完整鏈路,驗證技術可行性。代碼寫得很粗糙,配置全靠 .env 文件,賬號信息硬編碼,沒有數據庫,沒有調度,發佈完就結束了。但它能跑,這是最重要的。

同一天下午:重構前端,引入數據庫和調度

原型跑通後兩個小時,就做了一次大規模重構。靜態 HTML 換成了 React + Vite,引入了 SQLite 做持久化,加了 APScheduler 做定時調度,manager_service​ 和 goal_service 也在這次提交裏出現了。從提交記錄看,這次改動涉及 32 個文件、新增 6000 多行代碼,是整個項目體量最大的一次提交。

這種"先跑通再重構"的節奏是有意為之的。如果一開始就設計完整架構,很容易在不確定的地方過度設計。先有一個能跑的版本,再根據實際需要補充結構,反而更高效。

第一天傍晚:XHS 簽名逆向

發佈功能遇到了第一個硬骨頭——小紅書的請求籤名。upload_service​ 需要正確生成 x-s​、x-t​ 等簽名參數,否則請求會被拒絕。這部分涉及 JS 逆向,提交記錄裏能看到當時留下了大量的調試文件:xhs_xs.js​、xhs_xmns.js​、xhs_xs_deprecated.js​、xhs_xs_new.js​,還有一個 xhs_encrpty_test.py​ 用來驗證簽名結果。最終用 pyexecjs 在 Python 裏執行 JS 簽名邏輯,繞過了這個問題。

第二天:配置系統和 Docker

第二天的重點是讓項目"可部署"。配置從 .env 文件遷移到數據庫,前端新增了系統配置頁面,用户可以在 Web 界面裏填寫 API Key,不再需要手動編輯文件。同一天還做了 Docker 支持,Dockerfile 改為多階段構建,前端構建集成進去,最終產物是一個包含前後端的單容器鏡像。

這次改動還順手解決了一個日誌問題:middleware.py​ 裏的 logging.basicConfig() 會搶佔 root logger,導致文件日誌失效。這種問題在本地開發時完全感知不到,只有在 Docker 環境裏看日誌時才會發現什麼都沒寫進去。

第三天:提示詞 Agent 和日誌系統

prompt_agent​ 是在第三天加進來的。在這之前,圖片提示詞直接由 text_service 的 LLM 順帶生成,質量不穩定,有時候生成的提示詞太抽象,有時候風格和正文完全對不上。

加了 prompt_agent​ 之後,提示詞生成變成了一個獨立的步驟:預設 8 種模板,LLM 從中選擇最合適的再填充場景細節。這個改動同時修復了一個 bug——styles​ 參數在 xhs_agent.py​ 裏生成了但沒有傳遞給 image_service,導致所有圖片都用默認風格生成。這種 bug 在代碼裏很隱蔽,因為功能表面上是正常的,只是風格決策完全沒生效。

同一天還完善了日誌系統,終端只輸出 INFO,文件記錄 DEBUG,所有 AI 請求的完整提示詞和 LLM 原始響應都寫入文件日誌,方便排查問題。

第五天:參考圖片系統——最複雜的一次迭代

8d85f9d 這個提交是整個項目迭代過程中改動最集中的一次,涉及 22 個文件、新增 1500 多行代碼,在凌晨兩點多提交。這次把參考圖片系統從零搭起來:COS 存儲、GLM-4.6V 視覺識別、5 種分類體系、組模式管理,以及把參考圖標註和 COS URL 串聯進完整的生成鏈路。

最複雜的部分是"標註如何影響生圖"這條鏈路的設計。標註文字要傳給 text_service​ 影響風格決策,同時也要傳給 prompt_agent​ 融入提示詞,而 COS URL 要直接傳給圖片生成 API 做圖生圖參考。這三條路徑是獨立的,但都依賴同一份參考圖數據,需要在 goal_service 裏統一加載後分發給各個下游服務。

這次迭代還順帶做了數據庫增量遷移,因為新增了 image_groups​ 和 account_images​ 兩張表,同時給 scheduled_posts​ 加了 ref_image_ids 字段。之前做的自動遷移機制在這裏發揮了作用,不需要手動處理舊數據庫的兼容問題。

第六天:poster 串行生成,兩次提交解決一個問題

poster 風格的串行生成是在第六天加的,但有意思的是它被拆成了兩個提交,間隔只有三分鐘。第一個提交 8182262​ 實現了基本的串行邏輯:第一張生成後把結果 URL 追加到參考圖列表,後續圖片都用這個列表。第二個提交 83cd67f 緊接着做了一次 refactor:把"總管 AI 傳入的參考圖"和"第一張圖的錨定 URL"分開管理,不讓它們互相污染。

這兩個提交的間隔説明第一版實現完成後立刻發現了設計上的問題——兩組參考圖混在一起,邏輯不清晰,日誌也看不出參考圖的來源。三分鐘後直接重構,這種"寫完立刻改"的習慣在小項目裏其實很高效。

最後的修修補補

項目後期的提交基本都是修 bug 和細節打磨。話題標籤重複出現的問題(正文末尾手動拼接了 #話題,同時 topics 參數也會帶話題,導致發佈後出現兩遍)、圖片下載超時時間從 30s 調到 120s 再調到 240s(説明實際跑起來後發現生產環境的圖片下載比預期慢很多)、刪除運營目標時忘記同步取消定時任務(導致任務還會在後台觸發但找不到對應的目標記錄)。

這些 bug 大多是在實際使用中發現的,不是測試發現的——因為這個項目根本沒有寫測試。對於一個快速迭代驗證想法的工具項目來説,這是一個有意識的取捨。


總結

XHS Agent 的核心工程價值在於把多個 AI 能力和自動化流程編排成一條可靠的流水線。單個 AI 調用很容易實現,難的是讓整條鏈路在生產環境下穩定運行——失敗有重試,狀態有持久化,重啓能恢復,異常有通知。

user avatar
0 位用戶收藏了這個故事!

發佈 評論

Some HTML is okay.