基于Stigmergy的团队LLM wiki协作机制与FastAPI工程实践

Stigmergy 听起来像一个冷门词，但它可能是团队协作型 LLM wiki 最值得借鉴的设计原则。Andrej Karpathy 带动的 “LLM wiki” 讨论，通常指向一种个人化知识管理范式：一个人借助大语言模型持续提问、修订、链接页面，把零散笔记变成持续生长的维基。问题在于，知识一旦需要多人一起维护，这一个人玩得很转的模式会立刻撞上冲突、上下文丢失和信任问题。Stigmergy 恰好提供了一种不需要实时沟通、不依赖某个“主编”的协作机制：每个个体修改共享环境，后续个体根据环境的当前状态继续工作。这篇内容会围绕一个可运行的工程原型，解释如何把 Stigmergy 的原则落地成面向团队的 LLM wiki。

1. Stigmergy 是什么，为什么团队 LLM wiki 需要它

1.1 Stigmergy 的原始模型：个体留下环境痕迹，群体迭代出结构

Stigmergy 这个词源自生物学，最常被引用的例子是白蚁筑巢。白蚁个体并不掌握“巢穴设计图纸”，也不会开会讨论下一步怎么干。每只白蚁做的最多的事情，是在已有泥塔上放下一颗泥球，或者释放一点信息素。其他白蚁看到环境发生的改变后，会继续在同样的位置加工，泥塔越垒越高，最终形成复杂的巢穴结构。

这个机制有三个关键要素：环境状态可观察、个体行为会改变环境状态、后续个体根据新的环境状态行动。信息和意图不需要通过面对面交流传递，而是沉淀在环境里。软件工程中其实到处都有 Stigmergy：Git 提交历史、Pull Request 上的评论、CI 状态、代码里的 TODO 注释，都是环境痕迹。团队成员不直接“同步大脑”，而是通过对共享工件的修改来协作。

1.2 Karpathy 式单人 LLM wiki 的核心特征

所谓 Karpathy 式的 LLM wiki，可以理解成一种由大语言模型辅助维护的个人知识库。一个人把一篇篇笔记、文章、项目文档拆成相互链接的页面，用 LLM 来补充内容、调整结构、回答“这篇页面上一次写到这里，接下来应该补什么”。页面成为长期记忆单元，对话上下文只是短期工作台。

这种模式对单个人非常有效，因为所有知识状态都在同一个大脑里。用户不需要向外部解释背景，不需要说明自己为什么要改一段话，也不需要担心另一个人会误解页面上的某个措辞。LLM 只需要根据当前页面内容和一次对话指令，就能生成一版合理的修订稿。

但团队环境不具备这个前提。团队没有共享大脑，每个人的经验、假设、上下文都不同。如果仍然按单人模式操作，让多个成员各自和 LLM 对话，再手工粘贴到同一个目录里，结果几乎必然出现三类问题：同一段知识被改成多个版本、页面之间互相矛盾、没人知道某个结论是怎么得出的。

1.3 从一个人的知识库扩展到团队，问题立刻出现

把单人 LLM wiki 复制成团队版，第一个直接问题是并发覆盖。两个人同时基于旧版本编辑，后提交的人覆盖先提交的人。第二个问题是知识上下文丢失。A 写页面时知道自己为什么选择某个方案，但 B 和 LLM 看不到这层背景，修订时可能会把关键信息删掉。第三个问题是 LLM 的幻觉会被错误地当作权威内容。单人场景下，用户还能凭自己对知识的掌握判断是否正确；多人协作时，编辑者很可能并不熟悉页面涉及的专业领域。

这些问题靠“加一个登录页面”“加一个编辑锁”是解决不了的，因为问题本质上是协作模型没有设计好。系统需要让每个人和每个 Agent 都能看到当前知识状态、知道谁在哪些地方留过疑问、能追溯每一次修改的原因。

1.4 为什么 Stigmergy 恰好是解法

Stigmergy 的启示是：不要试图用实时通信去同步每个人，而是把知识库本身变成共享环境。页面是环境，版本历史是记录，评论、TODO、QUESTION、CONFLICT 标记就是环境中的痕迹。人或者 LLM Agent 不需要知道其他协作者在做什么，只需要查看当前页面上有哪些未解决的标记，基于当前版本继续贡献。

这样，先修改环境的人通过“痕迹”影响后来者，后来者又留下新痕迹。协作过程不需要强依赖记忆，也不需要一个中心协调者。LLM 在这个模型里的角色也从“个人写作助手”变成了“知识维护工人”：它读取当前页面、读取未解决标记、检索相关上下文，生成一版增量改动，最后由人确认发布。

下面的表格可以更直观地对比单人 LLM wiki 和团队 Stigmergy Wiki。

2. 团队版 LLM wiki 的数据结构和协作流程

2.1 三个核心实体：Page、Revision、Trace

在实现层面，团队版 LLM wiki 至少要包含三个核心实体。

Page 是当前知识状态。它有标题、正文、更新人、更新时间，以及一个用于乐观锁的版本号。Page 不保存历史，只保存当前有效内容。

Revision 是每一次正式发布的快照。每次页面更新都会产生一条新 Revision，记录版本号、作者、改动内容和备注。Revision 让系统可以回答“这个页面是怎么一步步变成现在这样的”，这是 Stigmergy 最基础的一层痕迹。

Trace 是页面上的协作痕迹。它可以是一条 QUESTION，比如“这里的幂等键为什么不用订单号？”；也可以是一条 CONFLICT，比如“A 方案和 B 方案在支付回调场景下矛盾”；还可以是一条 CONTEXT，比如“这段设计是基于银行渠道的 T+1 结算周期”。Trace 必须绑定页面版本，避免后来者用旧标记去干扰新内容。

Pages、Revisions、Traces 三者构成了一个可追溯的协作环境。LLM 在生成修订建议时，不应该只看页面正文，还应该读取当前未解决的 Trace，否则它不理解为什么页面会对某些问题保持特定写法。

2.2 一套支撑检索的向量索引

团队知识库通常会持续增长，页面之间会互相引用，LLM 单次输入又有上下文窗口限制。所以系统需要一套向量索引，按语义相似度检索相关页面和段落。

具体做法是：页面每次发布后，把内容切成若干 chunk，每个 chunk 通过 Embedding 模型转成向量，存入向量索引。当 LLM 要修订某个页面时，系统用“页面标题 + 用户指令 + 未解决标记”作为查询向量，检索最相关的几个历史片段和关联页面，再拼接进 Prompt。

向量索引解决的核心问题不是“找全所有资料”，而是“在有限上下文内找到最可能影响本次修改的知识片段”。生产环境可以接入 Qdrant、Weaviate 或 PostgreSQL 的 pgvector；学习环境可以用 SQLite 加一个简单的 cosine 相似度计算先跑通链路。

2.3 协作闭环怎么走通

一套基于 Stigmergy 的协作流程可以设计成六个步骤。

• 团队成员创建或编辑页面，系统生成新 Revision。
• 其他成员或 LLM Agent 在页面上添加 Trace 标记，比如提出问题、补充背景、标记冲突。
• 任一成员发起 LLM review，系统收集当前页面内容、未解决 Trace、相关页面片段。
• LLM 生成一版修订草案 draft，不直接覆盖当前页面。
• 人工确认 draft，并通过携带 expected_version 的发布接口更新页面。
• 页面更新后，向量索引同步刷新，未解决 Trace 保留，解决后的 Trace 标记为 resolved。

第 5 步是关键设计。不要让 LLM 拥有直接写页面的权限，至少不要让它在没有人确认的情况下覆盖线上内容。人的确认是防止幻觉污染的兜底机制。

2.4 技术选型和初始目录

学习环境可以使用 FastAPI 提供 API，SQLite 存储结构化数据，sentence-transformers 或兼容 OpenAI 协议的 Embedding 接口做向量化，Streamlit 或纯 curl 做验证。生产环境建议使用 PostgreSQL pgvector、独立的向量数据库、统一的模型网关，以及成熟的身份认证系统。

一个最小项目的目录可以这样组织。

stigmergy-wiki/ ├── app/ │ ├── main.py │ ├── db.py │ ├── llm.py │ ├── vector_index.py │ ├── stigmergy.py │ └── config.py ├── data/ ├── config.yaml └── requirements.txt

接下来从代码层面实现这个原型。

3. 用 FastAPI 和 SQLite 搭建最小可运行版本

3.1 初始化依赖和配置

先安装必要的依赖。requirements.txt 示例如下。

fastapi>=0.110 uvicorn[standard]>=0.29 pydantic>=2.6 PyYAML>=6.0 openai>=1.12 sentence-transformers>=2.2

sentence-transformers 用于本地向量化，openai 库用于调用兼容 OpenAI 协议的 REST 接口。实际项目建议先把依赖版本锁定，不要使用这种宽松版本范围部署到生产环境。

配置文件 config.yaml 可以这样设计。

database: path: "data/wiki.db" embedding: provider: "local" model: "BAAI/bge-small-zh-v1.5" chunk_size: 800 chunk_overlap: 100 llm: base_url: "http://localhost:11434/v1" api_key: "ollama" model: "qwen2.5:7b" temperature: 0.2

这里的 LLM 服务使用本地 Ollama 的 OpenAI 兼容地址，也可以换成团队自己的模型网关地址。不要把 API Key 写死在配置文件里，推荐通过环境变量注入。

3.2 建立数据库表

最小原型需要四张表：pages、revisions、traces、page_chunks。SQLite DDL 如下。

CREATE TABLE IF NOT EXISTS pages ( id TEXT PRIMARY KEY, title TEXT NOT NULL, content TEXT NOT NULL, version INTEGER NOT NULL DEFAULT 1, updated_by TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS revisions ( id INTEGER PRIMARY KEY AUTOINCREMENT, page_id TEXT NOT NULL, version INTEGER NOT NULL, content TEXT NOT NULL, diff TEXT NOT NULL DEFAULT '', author TEXT NOT NULL, note TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS traces ( id INTEGER PRIMARY KEY AUTOINCREMENT, page_id TEXT NOT NULL, page_version INTEGER NOT NULL, marker TEXT NOT NULL, message TEXT NOT NULL, author TEXT NOT NULL, resolved INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS page_chunks ( id INTEGER PRIMARY KEY AUTOINCREMENT, page_id TEXT NOT NULL, page_version INTEGER NOT NULL, chunk_index INTEGER NOT NULL, content TEXT NOT NULL, embedding TEXT NOT NULL );

Trace 表里的 marker 可以枚举为QUESTION、CONFLICT、CONTEXT、TODO。resolved 字段用于标记是否已经关闭。page_chunks 表在最小原型中把向量以 JSON 字符串存入 SQLite，方便验证逻辑；生产环境应当替换为真正的向量数据库。

3.3 实现页面、修订、痕迹 API

下面用 FastAPI 实现页面创建、查询、痕迹添加和发布接口。核心组件是 sqlite3 存储层。

先在 db.py 里实现连接和建表。

import sqlite3 from pathlib import Path DB_PATH = Path(__file__).resolve().parent.parent / "data" / "wiki.db" SCHEMA = """ -- 上面定义的 DDL """ def get_conn(): DB_PATH.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): conn = get_conn() try: conn.executescript(SCHEMA) conn.commit() finally: conn.close()

然后在 main.py 中实现 API。注意发布接口必须使用乐观锁，避免并发覆盖。

import uuid from datetime import datetime, timezone from typing import List, Optional from fastapi import FastAPI, HTTPException from pydantic import BaseModel from db import get_conn, init_db app = FastAPI(title="Stigmergy Wiki") class PageCreate(BaseModel): title: str content: str author: str class TraceCreate(BaseModel): marker: str message: str author: str class ReviewRequest(BaseModel): instruction: str class PublishRequest(BaseModel): content: str expected_version: int author: str note: str = "" @app.on_event("startup") def on_startup(): init_db() def now(): return datetime.now(timezone.utc).isoformat() @app.post("/pages") def create_page(payload: PageCreate): page_id = uuid.uuid4().hex ts = now() version = 1 conn = get_conn() try: conn.execute( "INSERT INTO pages (id, title, content, version, updated_by, updated_at) VALUES (?, ?, ?, ?, ?, ?)", (page_id, payload.title, payload.content, version, payload.author, ts), ) conn.execute( "INSERT INTO revisions (page_id, version, content, diff, author, note, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", (page_id, version, payload.content, "", payload.author, "initial", ts), ) conn.commit() finally: conn.close() return {"id": page_id, "version": version} @app.get("/pages/{page_id}") def get_page(page_id: str): conn = get_conn() try: row = conn.execute("SELECT * FROM pages WHERE id = ?", (page_id,)).fetchone() finally: conn.close() if row is None: raise HTTPException(status_code=404, detail="page not found") return dict(row) @app.post("/pages/{page_id}/traces") def add_trace(page_id: str, payload: TraceCreate): conn = get_conn() try: page = conn.execute("SELECT * FROM pages WHERE id = ?", (page_id,)).fetchone() if page is None: raise HTTPException(status_code=404, detail="page not found") ts = now() conn.execute( "INSERT INTO traces (page_id, page_version, marker, message, author, resolved, created_at) VALUES (?, ?, ?, ?, ?, 0, ?)", (page_id, page["version"], payload.marker, payload.message, payload.author, ts), ) conn.commit() trace_id = conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"] return {"id": trace_id, "page_version": page["version"]} finally: conn.close() @app.get("/pages/{page_id}/traces") def list_traces(page_id: str, unresolved_only: bool = True): conn = get_conn() try: if unresolved_only: rows = conn.execute( "SELECT * FROM traces WHERE page_id = ? AND resolved = 0 ORDER BY created_at", (page_id,), ).fetchall() else: rows = conn.execute( "SELECT * FROM traces WHERE page_id = ? ORDER BY created_at", (page_id,), ).fetchall() finally: conn.close() return [dict(r) for r in rows]

发布接口的关键在版本校验。

@app.put("/pages/{page_id}/publish") def publish_revision(page_id: str, payload: PublishRequest): conn = get_conn() try: page = conn.execute("SELECT * FROM pages WHERE id = ?", (page_id,)).fetchone() if page is None: raise HTTPException(status_code=404, detail="page not found") if page["version"] != payload.expected_version: raise HTTPException(status_code=409, detail="version conflict") new_version = page["version"] + 1 ts = now() conn.execute( "UPDATE pages SET content = ?, version = ?, updated_by = ?, updated_at = ? WHERE id = ?", (payload.content, new_version, payload.author, ts, page_id), ) conn.execute( "INSERT INTO revisions (page_id, version, content, diff, author, note, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", (page_id, new_version, payload.content, "", payload.author, payload.note, ts), ) conn.commit() return {"page_id": page_id, "new_version": new_version} except Exception: conn.rollback() raise finally: conn.close()

3.4 实现 LLM 修订生成器

LLM 修订生成器位于 llm.py。它负责读取当前页面、未解决 Trace、相关检索片段，并构造 Prompt。

import os from openai import OpenAI client = OpenAI( base_url=os.getenv("LLM_BASE_URL", "http://localhost:11434/v1"), api_key=os.getenv("LLM_API_KEY", "ollama"), ) def collect_traces_text(traces): lines = [] for t in traces: lines.append(f"- [{t['marker']}] {t['message']}（{t['author']}，页 v{t['page_version']}）") return "\n".join(lines) def build_review_prompt(page, traces_text, related_text, instruction): return f""" 你是一个团队知识库维护助手。你的任务是基于当前页面内容、协作痕迹和相关资料，生成一版修订稿。 要求： 1. 只输出修订后的完整页面内容，不要输出解释。 2. 尽可能保留原有有效信息。 3. 如果痕迹中包含问题或冲突，在修订稿中尽量回应。 4. 不要编造外部事实。 当前页面标题：{page['title']} 当前页面内容： {page['content']} 未解决协作痕迹： {traces_text} 相关检索片段： {related_text} 用户修订指令： {instruction} """ def generate_draft(page, traces, related_text, instruction): traces_text = collect_traces_text(traces) prompt = build_review_prompt(page, traces_text, related_text, instruction) response = client.chat.completions.create( model=os.getenv("LLM_MODEL", "qwen2.5:7b"), messages=[{"role": "user", "content": prompt}], temperature=0.2, ) return response.choices[0].message.content

这里将 temperature 设置得比较低，是为了让 LLM 更倾向保守修改，而不是天马行空地重写全文。实际项目还应该在调用中增加超时设置、重试策略和成本统计。

3.5 实现向量检索接口

向量索引模块负责把页面内容切分、向量化，并提供语义搜索接口。最小实现可以用 sentence-transformers 和 cosine 相似度。

import json import math from sentence_transformers import SentenceTransformer _model = None def get_model(): global _model if _model is None: _model = SentenceTransformer("BAAI/bge-small-zh-v1.5") return _model def embed_texts(texts): model = get_model() vectors = model.encode(texts, normalize_embeddings=True) return vectors.tolist() def cosine(a, b): return sum(x * y for x, y in zip(a, b)) def add_chunks(conn, page_id, page_version, content, chunk_size=800, overlap=100): chunks = [] start = 0 while start < len(content): chunks.append(content[start:start + chunk_size]) start += chunk_size - overlap vectors = embed_texts(chunks) for idx, (chunk, vec) in enumerate(zip(chunks, vectors)): conn.execute( "INSERT INTO page_chunks (page_id, page_version, chunk_index, content, embedding) VALUES (?, ?, ?, ?, ?)", (page_id, page_version, idx, chunk, json.dumps(vec)), ) def search(conn, query, top_k=3): qvec = embed_texts([query])[0] rows = conn.execute("SELECT * FROM page_chunks").fetchall() scored = [] for row in rows: vec = json.loads(row["embedding"]) score = cosine(qvec, vec) scored.append((score, row)) scored.sort(key=lambda x: x[0], reverse=True) return [dict(row) for _, row in scored[:top_k]]

学习环境直接全表扫描即可。生产环境数据量增大后，建议换成 pgvector 或 Qdrant，并让查询命中索引。

4. 嵌入、标记和冲突控制：细节决定 LLM 协作质量

4.1 页面分块和向量化时的关键参数

页面内容不是越大越好。过大的 chunk 会让向量语义被稀释，过小的 chunk 又会丢失上下文。常见参数建议如下。

不要以为向量越多越好。实际项目中，top_k 增加会直接增加 LLM 输入长度和成本，也可能引入噪声。建议先做一次小规模评测，记录每个查询命中的片段是否真的和问题相关。

4.2 如何把 Stigmergy 标记转换成 Prompt

Stigmergy 标记要发挥作用，必须能被 LLM 准确理解。最简单的方式是把未解决的 Trace 格式化成结构化的文本，放进 Prompt 中。

例如页面上有这样