WorkBuddy 企业实战：构建 AI 数字同事的完整架构与代码实现

第一代：FAQ 机器人     -> 关键词匹配，答不上就转人工
第二代：LLM 聊天助手   -> 能对话，但只能"说"，不能"做"
第三代：WorkBuddy      -> 能理解、能规划、能调用系统、能执行、能主动提醒

┌────────────────────────────────────────────┐
│ 1. 理解    理解自然语言、业务术语、上下文    │
│ 2. 规划    把模糊目标拆成可执行步骤          │
│ 3. 执行    调用企业系统 API / 工具           │
│ 4. 记忆    记住用户偏好、项目历史、组织知识  │
│ 5. 协作    与多人、多系统、多 WorkBuddy 协同 │
│ 6. 主动    定时任务、异常提醒、主动建议      │
└────────────────────────────────────────────┘

┌───────────────────────────────────────────────────────┐
│                      接入层                            │
│   Web / 移动端 / 飞书 / 钉钉 / Slack / API / Webhook    │
├───────────────────────────────────────────────────────┤
│                      身份层                            │
│   SSO / OIDC / SAML / 租户 / 角色 / 数据边界            │
├───────────────────────────────────────────────────────┤
│                      Agent 层                          │
│   规划器 / 执行器 / 反思器 / 工具选择 / 多智能体协作      │
├───────────────────────────────────────────────────────┤
│                      能力层                            │
│   工具注册表 / RAG / 记忆 / 工作流 / 定时任务            │
├───────────────────────────────────────────────────────┤
│                      集成层                            │
│   MCP / CRM / ERP / OA / 日历 / 邮件 / 工单 / 数据库     │
├───────────────────────────────────────────────────────┤
│                      数据层                            │
│   PostgreSQL / Redis / 向量库 / 对象存储 / KMS           │
├───────────────────────────────────────────────────────┤
│                      治理层                            │
│   权限 / 审计 / 限流 / 内容安全 / 成本 / 可观测性         │
└───────────────────────────────────────────────────────┘

用户目标
  -> 理解意图
  -> 检索上下文（记忆 + RAG）
  -> 制定计划
  -> 选择工具
  -> 执行工具
  -> 观察结果
  -> 反思：是否完成？
  -> 未完成：继续循环
  -> 完成：输出结果 + 更新记忆

workbuddy/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── db.py
│   ├── models.py
│   ├── tools.py
│   ├── memory.py
│   ├── rag.py
│   ├── agent.py
│   ├── llm.py
│   ├── security.py
│   ├── audit.py
│   ├── workflow.py
│   └── scheduler.py
├── tests/
│   └── test_workbuddy.py
├── requirements.txt
└── Dockerfile

pip install fastapi uvicorn sqlmodel pydantic pytest httpx numpy

fastapi
uvicorn[standard]
sqlmodel
pydantic
pytest
httpx
numpy

from datetime import datetime
from typing import Optional
from sqlmodel import SQLModel, Field


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(index=True, unique=True)
    display_name: str = ""
    department: str = ""
    role: str = "member"              # admin / manager / member / guest
    enabled: bool = True
    created_at: datetime = Field(default_factory=datetime.utcnow)


class Conversation(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(index=True)
    title: str = ""
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)


class Message(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    conversation_id: int = Field(index=True)
    role: str                          # user / assistant / tool / system
    content: str
    tool_name: Optional[str] = None
    created_at: datetime = Field(default_factory=datetime.utcnow)


class Memory(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(index=True)
    kind: str                          # preference / fact / project
    key: str
    value: str
    created_at: datetime = Field(default_factory=datetime.utcnow)


class Document(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    content: str
    department: str = ""
    tags: str = ""
    created_at: datetime = Field(default_factory=datetime.utcnow)


class ToolCallLog(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int
    conversation_id: int
    tool_name: str
    arguments: str
    result: str
    status: str = "ok"                 # ok / error / denied
    created_at: datetime = Field(default_factory=datetime.utcnow)


class ScheduledTask(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(index=True)
    name: str
    cron: str
    prompt: str
    enabled: bool = True
    last_run_at: Optional[datetime] = None
    created_at: datetime = Field(default_factory=datetime.utcnow)

from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List
from pydantic import BaseModel


@dataclass
class ToolSpec:
    name: str
    description: str
    parameters: Dict[str, Any]
    handler: Callable[..., Any]
    requires_roles: List[str] = field(default_factory=lambda: ["member"])
    risk_level: str = "low"           # low / medium / high
    requires_confirmation: bool = False


class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, ToolSpec] = {}

    def register(self, spec: ToolSpec):
        self._tools[spec.name] = spec

    def get(self, name: str) -> ToolSpec | None:
        return self._tools.get(name)

    def list_for_role(self, role: str) -> List[ToolSpec]:
        return [
            t for t in self._tools.values()
            if role in t.requires_roles
        ]

    def to_llm_schema(self, role: str) -> List[dict]:
        return [
            {
                "name": t.name,
                "description": t.description,
                "parameters": t.parameters,
            }
            for t in self.list_for_role(role)
        ]


registry = ToolRegistry()

def tool_calendar_create(user_id: int, title: str, start: str, duration_min: int = 60) -> dict:
    # 生产环境接入真实日历 API
    return {
        "event_id": f"evt_{user_id}_{abs(hash(title)) % 100000}",
        "title": title,
        "start": start,
        "duration_min": duration_min,
        "status": "created",
    }


def tool_ticket_create(user_id: int, title: str, priority: str = "medium") -> dict:
    return {
        "ticket_id": f"TK{abs(hash(title)) % 100000}",
        "title": title,
        "priority": priority,
        "status": "open",
    }


def tool_email_list_unread(user_id: int, limit: int = 10) -> dict:
    # 生产环境接入邮件系统
    return {
        "count": 3,
        "messages": [
            {"from": "client-a@example.com", "subject": "合同确认", "priority": "high"},
            {"from": "partner@example.com", "subject": "合作邀约", "priority": "medium"},
            {"from": "internal@example.com", "subject": "周报提醒", "priority": "low"},
        ][:limit],
    }


def tool_project_status(project_name: str) -> dict:
    return {
        "project": project_name,
        "progress": 0.62,
        "risks": ["关键路径延迟 3 天", "客户 SSO 对接未完成"],
        "owner": "alice",
    }


def tool_kb_search(query: str, top_k: int = 3) -> dict:
    from app.rag import search_documents
    return {"results": search_documents(query, top_k=top_k)}


def register_default_tools():
    registry.register(ToolSpec(
        name="calendar_create",
        description="创建一个日历事件",
        parameters={
            "type": "object",
            "properties": {
                "title": {"type": "string"},
                "start": {"type": "string", "description": "ISO8601 时间"},
                "duration_min": {"type": "integer", "default": 60},
            },
            "required": ["title", "start"],
        },
        handler=tool_calendar_create,
        requires_roles=["member", "manager", "admin"],
        risk_level="medium",
    ))

    registry.register(ToolSpec(
        name="ticket_create",
        description="创建一个工单",
        parameters={
            "type": "object",
            "properties": {
                "title": {"type": "string"},
                "priority": {"type": "string", "enum": ["low", "medium", "high", "urgent"]},
            },
            "required": ["title"],
        },
        handler=tool_ticket_create,
        requires_roles=["member", "manager", "admin"],
        risk_level="low",
    ))

    registry.register(ToolSpec(
        name="email_list_unread",
        description="列出当前用户未读邮件",
        parameters={
            "type": "object",
            "properties": {
                "limit": {"type": "integer", "default": 10},
            },
        },
        handler=tool_email_list_unread,
        requires_roles=["member", "manager", "admin"],
        risk_level="low",
    ))

    registry.register(ToolSpec(
        name="project_status",
        description="查询项目进度与风险",
        parameters={
            "type": "object",
            "properties": {
                "project_name": {"type": "string"},
            },
            "required": ["project_name"],
        },
        handler=tool_project_status,
        requires_roles=["member", "manager", "admin"],
        risk_level="low",
    ))

    registry.register(ToolSpec(
        name="kb_search",
        description="在企业知识库中检索文档",
        parameters={
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "top_k": {"type": "integer", "default": 3},
            },
            "required": ["query"],
        },
        handler=tool_kb_search,
        requires_roles=["member", "manager", "admin"],
        risk_level="low",
    ))


register_default_tools()

import json
from sqlmodel import Session
from app.models import ToolCallLog
from app.tools import registry, ToolSpec


class ToolExecutor:
    def __init__(self, db: Session, user, conversation_id: int):
        self.db = db
        self.user = user
        self.conversation_id = conversation_id

    def execute(self, tool_name: str, arguments: dict) -> dict:
        spec: ToolSpec | None = registry.get(tool_name)
        if not spec:
            return {"error": f"未知工具：{tool_name}"}

        if self.user.role not in spec.requires_roles:
            self._log(tool_name, arguments, "permission denied", "denied")
            return {"error": "无权限调用该工具"}

        try:
            result = spec.handler(user_id=self.user.id, **arguments)
            self._log(tool_name, arguments, result, "ok")
            return result
        except TypeError as e:
            self._log(tool_name, arguments, str(e), "error")
            return {"error": f"参数错误：{e}"}
        except Exception as e:
            self._log(tool_name, arguments, str(e), "error")
            return {"error": f"执行失败：{e}"}

    def _log(self, tool_name: str, args: dict, result, status: str):
        log = ToolCallLog(
            user_id=self.user.id,
            conversation_id=self.conversation_id,
            tool_name=tool_name,
            arguments=json.dumps(args, ensure_ascii=False),
            result=json.dumps(result, ensure_ascii=False, default=str),
            status=status,
        )
        self.db.add(log)
        self.db.commit()

from sqlmodel import Session, select
from app.models import Memory


def remember(db: Session, user_id: int, kind: str, key: str, value: str):
    existing = db.exec(
        select(Memory).where(
            Memory.user_id == user_id,
            Memory.kind == kind,
            Memory.key == key,
        )
    ).first()

    if existing:
        existing.value = value
        db.add(existing)
    else:
        db.add(Memory(user_id=user_id, kind=kind, key=key, value=value))
    db.commit()


def recall(db: Session, user_id: int, kind: str | None = None) -> dict:
    stmt = select(Memory).where(Memory.user_id == user_id)
    if kind:
        stmt = stmt.where(Memory.kind == kind)

    rows = db.exec(stmt).all()
    return {r.key: r.value for r in rows}


def build_memory_context(db: Session, user_id: int) -> str:
    memories = recall(db, user_id)
    if not memories:
        return ""

    lines = ["已知用户信息："]
    for k, v in memories.items():
        lines.append(f"- {k}: {v}")
    return "\n".join(lines)

remember(db, user_id=1, kind="preference", key="时区", value="Asia/Shanghai")
remember(db, user_id=1, kind="preference", key="会议时长", value="默认 45 分钟")
remember(db, user_id=1, kind="fact", key="负责项目", value="AI 客服系统")

import numpy as np
from sqlmodel import Session, select
from app.models import Document
from app.db import engine


def simple_embed(text: str, dim: int = 128) -> np.ndarray:
    """
    极简哈希嵌入，仅用于演示。
    生产环境请替换为 OpenAI / BGE / 本地嵌入模型。
    """
    vec = np.zeros(dim, dtype=np.float32)
    for token in text.lower().split():
        h = hash(token) % dim
        vec[h] += 1.0
    norm = np.linalg.norm(vec) + 1e-9
    return vec / norm


def index_document(db: Session, title: str, content: str, department: str = "", tags: str = ""):
    doc = Document(title=title, content=content, department=department, tags=tags)
    db.add(doc)
    db.commit()
    db.refresh(doc)
    return doc


def search_documents(query: str, top_k: int = 3) -> list:
    with Session(engine) as db:
        docs = db.exec(select(Document)).all()
    if not docs:
        return []

    q_vec = simple_embed(query)
    scored = []
    for d in docs:
        d_vec = simple_embed(d.title + " " + d.content)
        score = float(np.dot(q_vec, d_vec))
        scored.append((score, d))

    scored.sort(key=lambda x: x[0], reverse=True)
    return [
        {
            "id": d.id,
            "title": d.title,
            "snippet": d.content[:200],
            "score": round(score, 4),
        }
        for score, d in scored[:top_k]
    ]

from dataclasses import dataclass


@dataclass
class LLMDecision:
    kind: str                    # "tool" / "final"
    tool_name: str | None = None
    arguments: dict | None = None
    content: str | None = None


def mock_llm_decide(
    user_message: str,
    tool_schemas: list,
    history: list,
    memory_context: str,
) -> LLMDecision:
    """
    Mock LLM：用规则模拟决策，便于本地运行。
    生产环境替换为 OpenAI / Claude 的 function calling。
    """
    text = user_message.lower()

    if "未读邮件" in user_message or "邮件" in user_message:
        return LLMDecision(
            kind="tool",
            tool_name="email_list_unread",
            arguments={"limit": 5},
        )

    if "安排" in user_message and "会" in user_message:
        return LLMDecision(
            kind="tool",
            tool_name="calendar_create",
            arguments={
                "title": "团队评审会",
                "start": "2026-10-01T14:00:00",
                "duration_min": 45,
            },
        )

    if "工单" in user_message:
        return LLMDecision(
            kind="tool",
            tool_name="ticket_create",
            arguments={"title": user_message[:40], "priority": "high"},
        )

    if "项目" in user_message and ("进度" in user_message or "风险" in user_message):
        return LLMDecision(
            kind="tool",
            tool_name="project_status",
            arguments={"project_name": "AI 客服系统"},
        )

    if "知识库" in user_message or "制度" in user_message or "流程" in user_message:
        return LLMDecision(
            kind="tool",
            tool_name="kb_search",
            arguments={"query": user_message, "top_k": 3},
        )

    return LLMDecision(
        kind="final",
        content=f"收到你的消息：{user_message}。我可以帮你查邮件、排会议、查项目、建工单或检索知识库。",
    )

from sqlmodel import Session
from app.models import Conversation, Message, User
from app.llm import mock_llm_decide, LLMDecision
from app.tools import registry
from app.tools import ToolExecutor
from app.memory import build_memory_context, remember


MAX_STEPS = 5


class WorkBuddyAgent:
    def __init__(self, db: Session, user: User, conversation_id: int):
        self.db = db
        self.user = user
        self.conversation_id = conversation_id
        self.executor = ToolExecutor(db, user, conversation_id)

    def _history(self, limit: int = 20) -> list:
        rows = self.db.query(Message).filter(
            Message.conversation_id == self.conversation_id
        ).order_by(Message.id.desc()).limit(limit).all()
        return list(reversed(rows))

    def _save(self, role: str, content: str, tool_name: str | None = None):
        msg = Message(
            conversation_id=self.conversation_id,
            role=role,
            content=content,
            tool_name=tool_name,
        )
        self.db.add(msg)
        self.db.commit()
        self.db.refresh(msg)
        return msg

    def run(self, user_message: str) -> dict:
        self._save("user", user_message)

        memory_context = build_memory_context(self.db, self.user.id)
        tool_schemas = registry.to_llm_schema(self.user.role)

        steps = []

        for step in range(MAX_STEPS):
            decision: LLMDecision = mock_llm_decide(
                user_message=user_message,
                tool_schemas=tool_schemas,
                history=self._history(),
                memory_context=memory_context,
            )

            if decision.kind == "final":
                self._save("assistant", decision.content or "")
                return {
                    "reply": decision.content,
                    "steps": steps,
                    "conversation_id": self.conversation_id,
                }

            tool_name = decision.tool_name or ""
            args = decision.arguments or {}
            steps.append({"tool": tool_name, "arguments": args})

            result = self.executor.execute(tool_name, args)
            self._save("tool", str(result), tool_name=tool_name)

            # 简单反射：把工具结果拼进下一轮用户消息
            user_message = f"{user_message}\n[工具 {tool_name} 返回] {result}"

            # 记忆沉淀
            if tool_name == "calendar_create":
                remember(
                    self.db, self.user.id,
                    kind="fact", key="最近创建的会议",
                    value=result.get("title", ""),
                )

        final = "任务步骤过多，已停止。请拆分后再试。"
        self._save("assistant", final)
        return {"reply": final, "steps": steps, "conversation_id": self.conversation_id}

from fastapi import Header, HTTPException, Depends
from sqlmodel import Session, select
from app.db import get_session
from app.models import User

# 示例：生产环境使用 SSO / OIDC
API_KEYS = {
    "admin-key": "admin",
    "manager-key": "manager",
    "member-key": "member",
}


def get_current_user(
    x_api_key: str = Header(..., alias="X-API-Key"),
    session: Session = Depends(get_session),
) -> User:
    role = API_KEYS.get(x_api_key)
    if not role:
        raise HTTPException(status_code=401, detail="Invalid API Key")

    user = session.exec(
        select(User).where(User.username == role)
    ).first()
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    return user


def require_role(*roles):
    def checker(user: User = Depends(get_current_user)):
        if user.role not in roles:
            raise HTTPException(status_code=403, detail="Forbidden")
        return user
    return checker

from sqlmodel import Session, select
from app.models import ToolCallLog


def list_tool_calls(db: Session, user_id: int | None = None, limit: int = 200):
    stmt = select(ToolCallLog)
    if user_id:
        stmt = stmt.where(ToolCallLog.user_id == user_id)
    return db.exec(stmt.order_by(ToolCallLog.id.desc()).limit(limit)).all()

from sqlmodel import Session, select
from app.models import ScheduledTask, User
from app.agent import WorkBuddyAgent
from app.models import Conversation


def run_scheduled_task(db: Session, task: ScheduledTask) -> dict:
    user = db.get(User, task.user_id)
    if not user:
        return {"error": "user not found"}

    conv = Conversation(user_id=user.id, title=f"[定时] {task.name}")
    db.add(conv)
    db.commit()
    db.refresh(conv)

    agent = WorkBuddyAgent(db, user, conv.id)
    return agent.run(task.prompt)


def list_tasks(db: Session, user_id: int):
    return db.exec(
        select(ScheduledTask).where(ScheduledTask.user_id == user_id)
    ).all()

from datetime import datetime
from sqlmodel import Session, select
from app.models import ScheduledTask
from app.workflow import run_scheduled_task


def run_due_tasks(db: Session):
    """
    简化版调度：外部 cron 每分钟调用一次。
    生产环境用 APScheduler / Celery Beat / K8s CronJob。
    """
    tasks = db.exec(
        select(ScheduledTask).where(ScheduledTask.enabled == True)  # noqa: E712
    ).all()

    results = []
    for t in tasks:
        try:
            result = run_scheduled_task(db, t)
            t.last_run_at = datetime.utcnow()
            db.add(t)
            db.commit()
            results.append({"task_id": t.id, "result": result})
        except Exception as e:
            results.append({"task_id": t.id, "error": str(e)})
    return results

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from sqlmodel import Session

from app.db import init_db, get_session
from app.models import User, Conversation, Document, ScheduledTask
from app.security import get_current_user, require_role
from app.agent import WorkBuddyAgent
from app.audit import list_tool_calls
from app.rag import index_document
from app.tools import registry
from app.workflow import list_tasks

app = FastAPI(title="WorkBuddy - Enterprise Digital Coworker")


@app.on_event("startup")
def on_startup():
    init_db()


class ChatRequest(BaseModel):
    message: str
    conversation_id: int | None = None


class DocRequest(BaseModel):
    title: str
    content: str
    department: str = ""
    tags: str = ""


class TaskRequest(BaseModel):
    name: str
    cron: str
    prompt: str


@app.get("/health")
def health():
    return {"status": "ok"}


@app.get("/tools")
def list_tools(user: User = Depends(get_current_user)):
    return [
        {
            "name": t.name,
            "description": t.description,
            "risk_level": t.risk_level,
        }
        for t in registry.list_for_role(user.role)
    ]


@app.post("/chat")
def chat(
    payload: ChatRequest,
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    if payload.conversation_id:
        conv = session.get(Conversation, payload.conversation_id)
        if not conv or conv.user_id != user.id:
            raise HTTPException(status_code=404, detail="Conversation not found")
    else:
        conv = Conversation(user_id=user.id, title=payload.message[:30])
        session.add(conv)
        session.commit()
        session.refresh(conv)

    agent = WorkBuddyAgent(session, user, conv.id)
    return agent.run(payload.message)


@app.post("/knowledge")
def add_knowledge(
    payload: DocRequest,
    user: User = Depends(require_role("admin", "manager")),
    session: Session = Depends(get_session),
):
    doc = index_document(
        session,
        title=payload.title,
        content=payload.content,
        department=payload.department,
        tags=payload.tags,
    )
    return {"id": doc.id, "title": doc.title}


@app.post("/schedules")
def create_schedule(
    payload: TaskRequest,
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    task = ScheduledTask(
        user_id=user.id,
        name=payload.name,
        cron=payload.cron,
        prompt=payload.prompt,
    )
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


@app.get("/schedules")
def get_schedules(
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    return list_tasks(session, user.id)


@app.get("/audit/tool-calls")
def audit(
    user: User = Depends(require_role("admin", "manager")),
    session: Session = Depends(get_session),
):
    return list_tool_calls(session)

import os
import pytest
from fastapi.testclient import TestClient

os.environ["DATABASE_URL"] = "sqlite:///./test_workbuddy.db"

from app.main import app  # noqa: E402
from app.db import init_db, engine  # noqa: E402
from app.models import User  # noqa: E402
from sqlmodel import Session  # noqa: E402

client = TestClient(app)


def setup_module():
    init_db()
    with Session(engine) as db:
        for username, role in [
            ("admin", "admin"),
            ("manager", "manager"),
            ("member", "member"),
        ]:
            if not db.query(User).filter(User.username == username).first():
                db.add(User(username=username, display_name=username, role=role))
        db.commit()


def test_health():
    assert client.get("/health").json()["status"] == "ok"


def test_tools_list():
    r = client.get("/tools", headers={"X-API-Key": "member-key"})
    assert r.status_code == 200
    names = [t["name"] for t in r.json()]
    assert "email_list_unread" in names


def test_chat_email():
    r = client.post(
        "/chat",
        json={"message": "帮我看下未读邮件"},
        headers={"X-API-Key": "member-key"},
    )
    assert r.status_code == 200
    data = r.json()
    assert len(data["steps"]) >= 1
    assert data["steps"][0]["tool"] == "email_list_unread"


def test_chat_calendar():
    r = client.post(
        "/chat",
        json={"message": "帮我安排一个团队会"},
        headers={"X-API-Key": "member-key"},
    )
    assert r.status_code == 200
    assert r.json()["steps"][0]["tool"] == "calendar_create"


def test_chat_project():
    r = client.post(
        "/chat",
        json={"message": "查一下 AI 客服系统项目进度和风险"},
        headers={"X-API-Key": "manager-key"},
    )
    assert r.status_code == 200
    assert r.json()["steps"][0]["tool"] == "project_status"


def test_knowledge_and_search():
    r = client.post(
        "/knowledge",
        json={
            "title": "报销制度",
            "content": "员工出差报销需在 7 个工作日内提交，超过需部门经理审批。",
            "department": "财务",
        },
        headers={"X-API-Key": "admin-key"},
    )
    assert r.status_code == 200

    r = client.post(
        "/chat",
        json={"message": "查知识库报销流程"},
        headers={"X-API-Key": "member-key"},
    )
    assert r.status_code == 200
    assert r.json()["steps"][0]["tool"] == "kb_search"


def test_member_cannot_add_knowledge():
    r = client.post(
        "/knowledge",
        json={"title": "x", "content": "y"},
        headers={"X-API-Key": "member-key"},
    )
    assert r.status_code == 403


def test_schedule_create():
    r = client.post(
        "/schedules",
        json={
            "name": "每日业务摘要",
            "cron": "0 9 * * *",
            "prompt": "帮我总结未读邮件和项目风险",
        },
        headers={"X-API-Key": "member-key"},
    )
    assert r.status_code == 200
    assert r.json()["name"] == "每日业务摘要"

pytest -v

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app ./app

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

docker build -t workbuddy .
docker run -p 8000:8000 workbuddy

WorkBuddy = LLM + 工具 + 记忆 + 规划 + 权限 + 审计 + 企业集成

理解 -> 规划 -> 调用工具 -> 观察 -> 反思 -> 输出 -> 记忆沉淀 -> 主动提醒

workbuddy/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── db.py
│   ├── models.py
│   ├── tools.py
│   ├── memory.py
│   ├── rag.py
│   ├── agent.py
│   ├── llm.py
│   ├── security.py
│   ├── audit.py
│   ├── workflow.py
│   └── scheduler.py
├── tests/
│   └── test_workbuddy.py
├── requirements.txt
└── Dockerfile

pip install -r requirements.txt
pytest -v
uvicorn app.main:app --reload