Agent文档系统新手入门：从零搭建到生产环境部署

1次阅读

共计 3346 个字符，预计需要花费 9 分钟才能阅读完成。

在微服务架构下，传统文档管理系统（如 Wiki 或共享文件夹）暴露了三个明显缺陷：

版本碎片化：服务迭代时，不同团队的文档散落在多个平台，无法与 API 变更同步
权限颗粒度粗：基于目录的权限控制难以适应微服务间细粒度的访问需求
开发运维割裂：文档更新依赖人工维护，与 CI/CD 流程脱节

维度	Swagger	OpenAPI	Agent 文档系统
响应速度	中等（依赖 UI 渲染）	快（纯 JSON 格式）	极快（预编译缓存）
Schema 扩展性	有限（规范严格）	中等（支持 $ref）	高（自定义 DSL）
变更追溯	无内置版本控制	需外接 Git	内置版本快照
鉴权集成	Basic Auth	OAuth2.0	JWT+ABAC 模型

pip install fastapi uvicorn python-jose[cryptography] elasticsearch-dsl

# app/main.py
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from jose import JWTError, jwt

app = FastAPI()

# 模拟用户数据库
fake_users_db = {
    "johndoe": {"password": "secret"}
}

# JWT 配置
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

class Document(BaseModel):
    title: str
    content: str
    version: int = 1

@app.post("/token")
async def login(username: str, password: str):
    if username not in fake_users_db or fake_users_db[username]["password"] != password:
        raise HTTPException(status_code=400, detail="Incorrect credentials")

    token = jwt.encode({"sub": username}, SECRET_KEY, algorithm=ALGORITHM)
    return {"access_token": token}

@app.get("/documents/{doc_id}", response_model=Document)
async def read_document(doc_id: str, token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid authentication")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    # 实际应从数据库获取文档
    return Document(title="示例文档", content="这是文档内容")

Agent 文档采用类似 Git 的三级版本机制：

graph LR
    WorkingCopy -- 提交 --> StagingArea
    StagingArea -- 快照 --> VersionHistory
    VersionHistory -- 检出 --> WorkingCopy

工作区（Working Copy）：用户正在编辑的文档草稿
暂存区（Staging Area）：批量变更的中间状态
历史版本（Version History）：使用 SHA- 1 哈希存储不可变快照

from elasticsearch_dsl import Document as ESDocument, Text, Keyword

class ESArticle(ESDocument):
    title = Text(analyzer="ik_max_word")
    content = Text(analyzer="ik_smart")
    author = Keyword()
    version = Keyword()

    class Index:
        name = "agent_docs"
        settings = {
            "number_of_shards": 3,
            "number_of_replicas": 1
        }

优化要点：

对标题使用 ik_max_word 细粒度分词
正文内容使用 ik_smart 智能分词
版本号设为 Keyword 类型保证精确匹配

# app/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 主库写操作
write_engine = create_engine("postgresql://user:pwd@master-host:5432/db")
WriteSession = sessionmaker(bind=write_engine)

# 从库读操作
read_engine = create_engine("postgresql://user:pwd@replica-host:5432/db")
ReadSession = sessionmaker(bind=read_engine)

错误写法：

# 伪代码：每次循环都查询数据库
for doc in document_list:
    author = get_author(doc.author_id)  # 产生 N 次查询

正确方案：

# 一次性预加载关联数据
authors = {a.id: a for a in batch_get_authors(author_ids)}
for doc in document_list:
    author = authors[doc.author_id]

# app/middleware.py
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

# 不同端点差异化限流
@app.get("/search")
@limiter.limit("100/minute")
async def search_documents():
    return {...}

@app.post("/upload")
@limiter.limit("10/minute")
async def upload_document():
    return {...}

创建基础文档服务（复用前文代码）

集成 WebSocket 实现实时通信

from fastapi import WebSocket

@app.websocket("/ws/{doc_id}")
async def doc_websocket(websocket: WebSocket, doc_id: str):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        # 广播给所有连接的客户端
        for client in active_clients:
            await client.send_text(data)