WorkBuddy 零基础实战：从零构建你的 AI 工作助手

原创

资源大佬 jzit-top

发布于 2026-09-19 17:48:39

1390

摘要

WorkBuddy 不是一个具体的商业产品，而是一种产品形态——嵌入日常工作流的 AI 助手。它不追求"什么都能聊"，而是专注解决几个高频办公场景：任务管理、日程安排、邮件起草、文档摘要。本文从零基础出发，用 Python 构建一个可运行的 WorkBuddy，覆盖 Agent 循环、工具注册、SQLite 持久化、LLM 接入与 Mock 模式、CLI 交互与 Web API。读完你就能拥有一个属于自己的、能真正干活的工作助手。

一、WorkBuddy 是什么

先厘清概念。市面上有各种"AI 助手"，但大多数停留在"聊天框"层面：你问它答，聊完即忘，无法真正改变你的工作流。

WorkBuddy 的定位不同。它是一个能执行动作的助手：

类型	能力	局限
聊天机器人	问答、生成文本	不能操作数据，聊完即忘
RAG 问答	基于知识库回答	只读，不能写入
WorkBuddy	读写任务、日程、笔记，执行多步操作	需要工程实现

WorkBuddy 的核心是工具调用（Tool Calling）：用户说"帮我把明天的会议挪到下午三点"，助手理解意图后，调用 reschedule_event 工具，真正修改数据库里的日程。

二、零基础需要理解的三件事

在写代码前，只需要理解三个概念：

2.1 大语言模型（LLM）

LLM 是"会说话的推理引擎"。你给它一段文字，它给你回复。它本身不能查数据库、不能发邮件，只能生成文本。

2.2 工具调用

我们告诉 LLM："你有这些工具可用：add_task、list_tasks……" 当用户提出需求时，LLM 不直接回答，而是输出一个结构化请求：

{"tool": "add_task", "arguments": {"title": "写周报", "due": "2026-09-20"}}

我们的程序解析这个请求，执行 add_task 函数，把结果返回给 LLM，LLM 再生成最终回复。

2.3 Agent 循环

一次工具调用可能不够。用户说"整理我今天的任务并按优先级排序"，可能需要：查询任务 → 排序 → 更新顺序 → 汇报。这个"推理 → 调用 → 观察 → 再推理"的循环，就是 Agent 循环。

理解了这三点，剩下的就是工程。

三、架构设计

┌─────────────────────────────────────┐
│  用户界面（CLI / Web / 微信 / 飞书）   │
├─────────────────────────────────────┤
│  Agent 循环                          │
│  推理 → 工具调用 → 执行 → 观察 → 循环  │
├─────────────────────────────────────┤
│  工具注册表                           │
│  task / event / note / email         │
├─────────────────────────────────────┤
│  LLM 客户端（OpenAI / 本地 / Mock）    │
├─────────────────────────────────────┤
│  SQLite 持久化                        │
└─────────────────────────────────────┘

设计原则：

工具纯函数化：每个工具只做一件事，输入输出明确。
持久化与逻辑分离：工具通过 storage 层读写，不直接碰数据库。
Mock 优先：没有 API Key 也能跑，方便学习和测试。
可观测：每次工具调用都记录，方便调试。

四、项目结构

workbuddy/
├── workbuddy/
│   ├── __init__.py
│   ├── storage.py       # SQLite 持久化
│   ├── llm.py           # LLM 客户端（含 Mock）
│   ├── tools.py         # 工具注册表与实现
│   ├── agent.py         # Agent 循环
│   └── cli.py           # 命令行界面
├── requirements.txt
└── README.md

五、代码实现

5.1 requirements.txt

httpx>=0.27.0
pydantic>=2.5.0

只依赖两个库。httpx 用于调用 LLM API，pydantic 用于数据校验。SQLite 是 Python 标准库自带。

5.2 持久化层：storage.py

"""SQLite 持久化层：任务、日程、笔记。"""

import sqlite3
from contextlib import contextmanager
from datetime import datetime
from pathlib import Path
from typing import Any


class Storage:
    def __init__(self, db_path: str = "workbuddy.db"):
        self.db_path = db_path
        self._init_schema()

    @contextmanager
    def _conn(self):
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
            conn.commit()
        finally:
            conn.close()

    def _init_schema(self):
        with self._conn() as conn:
            conn.executescript("""
                CREATE TABLE IF NOT EXISTS tasks (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    title TEXT NOT NULL,
                    due TEXT,
                    priority TEXT DEFAULT 'medium',
                    status TEXT DEFAULT 'open',
                    created_at TEXT DEFAULT CURRENT_TIMESTAMP
                );

                CREATE TABLE IF NOT EXISTS events (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    title TEXT NOT NULL,
                    start_at TEXT NOT NULL,
                    end_at TEXT,
                    location TEXT,
                    created_at TEXT DEFAULT CURRENT_TIMESTAMP
                );

                CREATE TABLE IF NOT EXISTS notes (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    title TEXT,
                    content TEXT NOT NULL,
                    tags TEXT DEFAULT '',
                    created_at TEXT DEFAULT CURRENT_TIMESTAMP
                );
            """)

    # ---------- 任务 ----------
    def add_task(self, title: str, due: str | None = None,
                 priority: str = "medium") -> dict[str, Any]:
        with self._conn() as conn:
            cur = conn.execute(
                "INSERT INTO tasks (title, due, priority) VALUES (?, ?, ?)",
                (title, due, priority),
            )
            task_id = cur.lastrowid
        return self.get_task(task_id)

    def get_task(self, task_id: int) -> dict[str, Any] | None:
        with self._conn() as conn:
            row = conn.execute(
                "SELECT * FROM tasks WHERE id = ?", (task_id,)
            ).fetchone()
        return dict(row) if row else None

    def list_tasks(self, status: str = "open") -> list[dict[str, Any]]:
        with self._conn() as conn:
            if status == "all":
                rows = conn.execute(
                    "SELECT * FROM tasks ORDER BY "
                    "CASE priority WHEN 'high' THEN 1 "
                    "WHEN 'medium' THEN 2 ELSE 3 END, id"
                ).fetchall()
            else:
                rows = conn.execute(
                    "SELECT * FROM tasks WHERE status = ? ORDER BY "
                    "CASE priority WHEN 'high' THEN 1 "
                    "WHEN 'medium' THEN 2 ELSE 3 END, id",
                    (status,),
                ).fetchall()
        return [dict(r) for r in rows]

    def complete_task(self, task_id: int) -> dict[str, Any] | None:
        with self._conn() as conn:
            conn.execute(
                "UPDATE tasks SET status = 'done' WHERE id = ?", (task_id,)
            )
        return self.get_task(task_id)

    def delete_task(self, task_id: int) -> bool:
        with self._conn() as conn:
            cur = conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
        return cur.rowcount > 0

    # ---------- 日程 ----------
    def add_event(self, title: str, start_at: str,
                  end_at: str | None = None,
                  location: str | None = None) -> dict[str, Any]:
        with self._conn() as conn:
            cur = conn.execute(
                "INSERT INTO events (title, start_at, end_at, location) "
                "VALUES (?, ?, ?, ?)",
                (title, start_at, end_at, location),
            )
            event_id = cur.lastrowid
        return self.get_event(event_id)

    def get_event(self, event_id: int) -> dict[str, Any] | None:
        with self._conn() as conn:
            row = conn.execute(
                "SELECT * FROM events WHERE id = ?", (event_id,)
            ).fetchone()
        return dict(row) if row else None

    def list_events(self, date: str | None = None) -> list[dict[str, Any]]:
        with self._conn() as conn:
            if date:
                rows = conn.execute(
                    "SELECT * FROM events WHERE start_at LIKE ? "
                    "ORDER BY start_at",
                    (f"{date}%",),
                ).fetchall()
            else:
                rows = conn.execute(
                    "SELECT * FROM events ORDER BY start_at"
                ).fetchall()
        return [dict(r) for r in rows]

    def reschedule_event(self, event_id: int,
                         new_start: str,
                         new_end: str | None = None) -> dict[str, Any] | None:
        with self._conn() as conn:
            if new_end:
                conn.execute(
                    "UPDATE events SET start_at = ?, end_at = ? WHERE id = ?",
                    (new_start, new_end, event_id),
                )
            else:
                conn.execute(
                    "UPDATE events SET start_at = ? WHERE id = ?",
                    (new_start, event_id),
                )
        return self.get_event(event_id)

    # ---------- 笔记 ----------
    def add_note(self, content: str, title: str | None = None,
                 tags: str = "") -> dict[str, Any]:
        with self._conn() as conn:
            cur = conn.execute(
                "INSERT INTO notes (title, content, tags) VALUES (?, ?, ?)",
                (title, content, tags),
            )
            note_id = cur.lastrowid
        return self.get_note(note_id)

    def get_note(self, note_id: int) -> dict[str, Any] | None:
        with self._conn() as conn:
            row = conn.execute(
                "SELECT * FROM notes WHERE id = ?", (note_id,)
            ).fetchone()
        return dict(row) if row else None

    def search_notes(self, keyword: str) -> list[dict[str, Any]]:
        with self._conn() as conn:
            rows = conn.execute(
                "SELECT * FROM notes WHERE content LIKE ? OR title LIKE ? "
                "ORDER BY id DESC LIMIT 20",
                (f"%{keyword}%", f"%{keyword}%"),
            ).fetchall()
        return [dict(r) for r in rows]

5.3 LLM 客户端：llm.py

支持三种模式：Mock（无 Key）、OpenAI 兼容 API、本地 Ollama。

"""LLM 客户端：支持 Mock / OpenAI 兼容 / Ollama。"""

import json
import os
from typing import Any

import httpx


class LLMClient:
    def __init__(self, mode: str = "mock",
                 base_url: str = "https://api.openai.com/v1",
                 api_key: str | None = None,
                 model: str = "gpt-4o-mini"):
        self.mode = mode
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key or os.getenv("OPENAI_API_KEY", "")
        self.model = model

    def chat(self, messages: list[dict],
             tools: list[dict] | None = None) -> dict:
        """返回 {'content': str, 'tool_calls': [...]}"""
        if self.mode == "mock":
            return self._mock(messages, tools)
        if self.mode == "ollama":
            return self._ollama(messages, tools)
        return self._openai(messages, tools)

    # ---------- OpenAI 兼容 ----------
    def _openai(self, messages: list[dict],
                tools: list[dict] | None) -> dict:
        payload: dict[str, Any] = {
            "model": self.model,
            "messages": messages,
            "temperature": 0.1,
        }
        if tools:
            payload["tools"] = tools
            payload["tool_choice"] = "auto"

        headers = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"

        with httpx.Client(timeout=60) as client:
            resp = client.post(
                f"{self.base_url}/chat/completions",
                json=payload, headers=headers,
            )
            resp.raise_for_status()
            data = resp.json()

        msg = data["choices"][0]["message"]
        return {
            "content": msg.get("content") or "",
            "tool_calls": msg.get("tool_calls") or [],
        }

    # ---------- Ollama ----------
    def _ollama(self, messages: list[dict],
                tools: list[dict] | None) -> dict:
        payload: dict[str, Any] = {
            "model": self.model,
            "messages": messages,
            "stream": False,
        }
        if tools:
            payload["tools"] = tools

        with httpx.Client(timeout=120) as client:
            resp = client.post(
                f"{self.base_url}/api/chat", json=payload,
            )
            resp.raise_for_status()
            data = resp.json()

        msg = data.get("message", {})
        tool_calls = []
        for tc in msg.get("tool_calls", []) or []:
            fn = tc.get("function", {})
            tool_calls.append({
                "id": tc.get("id", "call_0"),
                "function": {
                    "name": fn.get("name", ""),
                    "arguments": json.dumps(fn.get("arguments", {})),
                },
            })
        return {"content": msg.get("content") or "",
                "tool_calls": tool_calls}

    # ---------- Mock ----------
    def _mock(self, messages: list[dict],
              tools: list[dict] | None) -> dict:
        """无 API Key 时的本地模拟：用规则匹配用户意图。"""
        user_text = ""
        for m in reversed(messages):
            if m.get("role") == "user":
                user_text = m.get("content", "")
                break

        text = user_text.lower()

        # 简单关键词路由
        if any(k in user_text for k in ["添加任务", "新建任务", "记个任务"]):
            title = user_text.split("任务", 1)[-1].strip("：: ") or "未命名任务"
            return {
                "content": "",
                "tool_calls": [{
                    "id": "call_mock_1",
                    "function": {
                        "name": "add_task",
                        "arguments": json.dumps(
                            {"title": title, "priority": "medium"}
                        ),
                    },
                }],
            }

        if any(k in user_text for k in ["列出任务", "看任务", "所有任务"]):
            return {
                "content": "",
                "tool_calls": [{
                    "id": "call_mock_2",
                    "function": {
                        "name": "list_tasks",
                        "arguments": json.dumps({"status": "open"}),
                    },
                }],
            }

        if any(k in user_text for k in ["完成", "做完"]):
            import re
            m = re.search(r"(\d+)", user_text)
            if m:
                return {
                    "content": "",
                    "tool_calls": [{
                        "id": "call_mock_3",
                        "function": {
                            "name": "complete_task",
                            "arguments": json.dumps(
                                {"task_id": int(m.group(1))}
                            ),
                        },
                    }],
                }

        if any(k in user_text for k in ["安排", "日程", "会议"]):
            return {
                "content": "",
                "tool_calls": [{
                    "id": "call_mock_4",
                    "function": {
                        "name": "list_events",
                        "arguments": json.dumps({}),
                    },
                }],
            }

        if any(k in user_text for k in ["记笔记", "保存笔记"]):
            content = user_text.split("笔记", 1)[-1].strip("：: ")
            return {
                "content": "",
                "tool_calls": [{
                    "id": "call_mock_5",
                    "function": {
                        "name": "add_note",
                        "arguments": json.dumps({"content": content}),
                    },
                }],
            }

        # 默认直接回答
        return {
            "content": (
                "我是 WorkBuddy，可以帮你管理任务、日程和笔记。"
                "试试说：帮我添加任务：写周报；列出所有任务；"
                "把任务 1 标记完成。"
            ),
            "tool_calls": [],
        }

5.4 工具注册表：tools.py

"""工具注册表：定义 Agent 可调用的所有工具。"""

import json
from typing import Any, Callable

from .storage import Storage


class Tool:
    def __init__(self, name: str, description: str,
                 parameters: dict, func: Callable):
        self.name = name
        self.description = description
        self.parameters = parameters
        self.func = func

    def run(self, arguments: dict) -> Any:
        return self.func(**arguments)

    def to_openai_schema(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            },
        }


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

    def register(self, tool: Tool):
        self._tools[tool.name] = tool

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

    def schemas(self) -> list[dict]:
        return [t.to_openai_schema() for t in self._tools.values()]


def build_registry(storage: Storage) -> ToolRegistry:
    reg = ToolRegistry()

    reg.register(Tool(
        name="add_task",
        description="添加一个待办任务",
        parameters={
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "任务标题"},
                "due": {"type": "string", "description": "截止日期 YYYY-MM-DD"},
                "priority": {
                    "type": "string",
                    "enum": ["low", "medium", "high"],
                    "description": "优先级",
                },
            },
            "required": ["title"],
        },
        func=lambda title, due=None, priority="medium":
            storage.add_task(title, due, priority),
    ))

    reg.register(Tool(
        name="list_tasks",
        description="列出任务",
        parameters={
            "type": "object",
            "properties": {
                "status": {
                    "type": "string",
                    "enum": ["open", "done", "all"],
                    "description": "筛选状态",
                },
            },
        },
        func=lambda status="open": storage.list_tasks(status),
    ))

    reg.register(Tool(
        name="complete_task",
        description="把任务标记为已完成",
        parameters={
            "type": "object",
            "properties": {
                "task_id": {"type": "integer", "description": "任务 ID"},
            },
            "required": ["task_id"],
        },
        func=lambda task_id: storage.complete_task(task_id),
    ))

    reg.register(Tool(
        name="delete_task",
        description="删除任务",
        parameters={
            "type": "object",
            "properties": {
                "task_id": {"type": "integer"},
            },
            "required": ["task_id"],
        },
        func=lambda task_id: {"deleted": storage.delete_task(task_id)},
    ))

    reg.register(Tool(
        name="add_event",
        description="添加日程",
        parameters={
            "type": "object",
            "properties": {
                "title": {"type": "string"},
                "start_at": {"type": "string",
                             "description": "开始时间 YYYY-MM-DD HH:MM"},
                "end_at": {"type": "string"},
                "location": {"type": "string"},
            },
            "required": ["title", "start_at"],
        },
        func=lambda title, start_at, end_at=None, location=None:
            storage.add_event(title, start_at, end_at, location),
    ))

    reg.register(Tool(
        name="list_events",
        description="列出日程",
        parameters={
            "type": "object",
            "properties": {
                "date": {"type": "string",
                         "description": "日期 YYYY-MM-DD，可选"},
            },
        },
        func=lambda date=None: storage.list_events(date),
    ))

    reg.register(Tool(
        name="reschedule_event",
        description="修改日程时间",
        parameters={
            "type": "object",
            "properties": {
                "event_id": {"type": "integer"},
                "new_start": {"type": "string"},
                "new_end": {"type": "string"},
            },
            "required": ["event_id", "new_start"],
        },
        func=lambda event_id, new_start, new_end=None:
            storage.reschedule_event(event_id, new_start, new_end),
    ))

    reg.register(Tool(
        name="add_note",
        description="保存笔记",
        parameters={
            "type": "object",
            "properties": {
                "content": {"type": "string"},
                "title": {"type": "string"},
                "tags": {"type": "string"},
            },
            "required": ["content"],
        },
        func=lambda content, title=None, tags="":
            storage.add_note(content, title, tags),
    ))

    reg.register(Tool(
        name="search_notes",
        description="搜索笔记",
        parameters={
            "type": "object",
            "properties": {
                "keyword": {"type": "string"},
            },
            "required": ["keyword"],
        },
        func=lambda keyword: storage.search_notes(keyword),
    ))

    return reg

5.5 Agent 循环：agent.py

"""Agent 循环：推理 → 工具调用 → 执行 → 观察 → 循环。"""

import json
from typing import Any

from .llm import LLMClient
from .tools import ToolRegistry


SYSTEM_PROMPT = """你是 WorkBuddy，一个工作助手。
你可以调用工具来管理用户的任务、日程和笔记。

规则：
1. 用户提出需求时，优先调用工具完成，而不是只给建议。
2. 工具调用后，基于结果用简洁中文回复用户。
3. 不确定用户意图时，先询问而不是猜测。
4. 涉及删除操作时，先向用户确认。
"""


class Agent:
    def __init__(self, llm: LLMClient, registry: ToolRegistry,
                 max_rounds: int = 5):
        self.llm = llm
        self.registry = registry
        self.max_rounds = max_rounds
        self.history: list[dict] = [
            {"role": "system", "content": SYSTEM_PROMPT}
        ]

    def reset(self):
        self.history = [{"role": "system", "content": SYSTEM_PROMPT}]

    def chat(self, user_message: str) -> str:
        self.history.append({"role": "user", "content": user_message})

        for _ in range(self.max_rounds):
            response = self.llm.chat(
                messages=self.history,
                tools=self.registry.schemas(),
            )
            content = response["content"]
            tool_calls = response["tool_calls"]

            # 无工具调用 → 最终回答
            if not tool_calls:
                self.history.append(
                    {"role": "assistant", "content": content}
                )
                return content or "（无回复）"

            # 记录 assistant 的工具调用意图
            assistant_msg: dict[str, Any] = {
                "role": "assistant",
                "content": content or "",
                "tool_calls": tool_calls,
            }
            self.history.append(assistant_msg)

            # 执行每个工具
            for call in tool_calls:
                result = self._execute_tool(call)
                self.history.append({
                    "role": "tool",
                    "tool_call_id": call.get("id", ""),
                    "content": json.dumps(result, ensure_ascii=False),
                })

        return "（已达到工具调用上限，请简化你的请求）"

    def _execute_tool(self, call: dict) -> Any:
        fn = call.get("function", {})
        name = fn.get("name", "")
        raw_args = fn.get("arguments", "{}")

        try:
            args = json.loads(raw_args) if isinstance(raw_args, str) else raw_args
        except json.JSONDecodeError:
            return {"error": f"参数解析失败: {raw_args}"}

        tool = self.registry.get(name)
        if not tool:
            return {"error": f"未知工具: {name}"}

        try:
            return tool.run(args)
        except Exception as e:
            return {"error": f"{type(e).__name__}: {e}"}

5.6 CLI 界面：cli.py

"""命令行界面：交互式对话。"""

import os
import sys

from .agent import Agent
from .llm import LLMClient
from .storage import Storage
from .tools import build_registry


def main():
    storage = Storage(os.getenv("WORKBUDDY_DB", "workbuddy.db"))
    registry = build_registry(storage)

    # 自动选择模式
    mode = os.getenv("WORKBUDDY_MODE", "")
    if not mode:
        mode = "openai" if os.getenv("OPENAI_API_KEY") else "mock"

    llm = LLMClient(
        mode=mode,
        base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
        api_key=os.getenv("OPENAI_API_KEY"),
        model=os.getenv("WORKBUDDY_MODEL", "gpt-4o-mini"),
    )

    agent = Agent(llm, registry)

    print("=" * 50)
    print("  WorkBuddy — 你的 AI 工作助手")
    print(f"  模式: {mode}")
    print("  输入 /help 查看示例，/reset 清空上下文，/quit 退出")
    print("=" * 50)

    while True:
        try:
            user_input = input("\n你 > ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n再见！")
            break

        if not user_input:
            continue
        if user_input in ("/quit", "/exit", "/q"):
            print("再见！")
            break
        if user_input == "/reset":
            agent.reset()
            print("上下文已重置。")
            continue
        if user_input == "/help":
            print("""
示例指令：
  帮我添加任务：写周报，优先级高
  列出所有任务
  把任务 1 标记完成
  添加日程：明天 15:00 产品评审
  列出日程
  记笔记：今天学到 SQLite 的 row_factory 用法
  搜索笔记：SQLite
            """.strip())
            continue

        try:
            reply = agent.chat(user_input)
        except Exception as e:
            print(f"\n[错误] {type(e).__name__}: {e}")
            continue

        print(f"\nWorkBuddy > {reply}")


if __name__ == "__main__":
    main()

5.7 入口文件：init.py

from .agent import Agent
from .llm import LLMClient
from .storage import Storage
from .tools import ToolRegistry, build_registry

__all__ = [
    "Agent", "LLMClient", "Storage",
    "ToolRegistry", "build_registry",
]
__version__ = "0.1.0"

六、运行与测试

6.1 安装

mkdir workbuddy && cd workbuddy
# 按上面结构创建文件
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install -r requirements.txt

6.2 运行（Mock 模式，无需 API Key）

python -m workbuddy.cli

输出示例：

==================================================
  WorkBuddy — 你的 AI 工作助手
  模式: mock
==================================================

你 > 帮我添加任务：写周报
WorkBuddy > 已为你添加任务《写周报》，优先级 medium。

你 > 列出所有任务
WorkBuddy > 当前有 1 个待办：1. 写周报 [medium]

你 > 把任务 1 标记完成
WorkBuddy > 任务 1《写周报》已标记为完成。

6.3 运行（接入真实 LLM）

export OPENAI_API_KEY=sk-xxxx
export WORKBUDDY_MODEL=gpt-4o-mini
python -m workbuddy.cli

接入本地 Ollama：

export WORKBUDDY_MODE=ollama
export OPENAI_BASE_URL=http://localhost:11434
export WORKBUDDY_MODEL=qwen2.5:7b
python -m workbuddy.cli

七、进阶：Web API 版本

CLI 适合个人使用。要多人共享，加一个 FastAPI 层。

7.1 安装额外依赖

pip install fastapi uvicorn

7.2 新增 web.py

"""WorkBuddy Web API。"""

import os

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from .agent import Agent
from .llm import LLMClient
from .storage import Storage
from .tools import build_registry

app = FastAPI(title="WorkBuddy API")

_storage = Storage(os.getenv("WORKBUDDY_DB", "workbuddy.db"))
_registry = build_registry(_storage)
_llm = LLMClient(
    mode=os.getenv("WORKBUDDY_MODE", "mock"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    api_key=os.getenv("OPENAI_API_KEY"),
    model=os.getenv("WORKBUDDY_MODEL", "gpt-4o-mini"),
)

# 简易多会话：生产环境应用 Redis 存储
_sessions: dict[str, Agent] = {}


class ChatRequest(BaseModel):
    session_id: str = "default"
    message: str


class ChatResponse(BaseModel):
    session_id: str
    reply: str


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


@app.post("/chat", response_model=ChatResponse)
def chat(req: ChatRequest):
    if not req.message.strip():
        raise HTTPException(400, "message is empty")

    agent = _sessions.get(req.session_id)
    if agent is None:
        agent = Agent(_llm, _registry)
        _sessions[req.session_id] = agent

    try:
        reply = agent.chat(req.message)
    except Exception as e:
        raise HTTPException(500, f"{type(e).__name__}: {e}")

    return ChatResponse(session_id=req.session_id, reply=reply)


@app.post("/sessions/{session_id}/reset")
def reset(session_id: str):
    agent = _sessions.get(session_id)
    if agent:
        agent.reset()
    return {"status": "reset"}


@app.get("/tasks")
def tasks(status: str = "open"):
    return _storage.list_tasks(status)


@app.get("/events")
def events(date: str | None = None):
    return _storage.list_events(date)

7.3 启动

uvicorn workbuddy.web:app --reload --port 8000

7.4 调用

curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"session_id":"u1","message":"帮我添加任务：写周报"}'

curl http://localhost:8000/tasks

八、从 Demo 到生产的 Checklist

当前实现是一个可运行的最小版本，生产化还需要：

□ 鉴权：API Key、OAuth、多用户隔离
□ 持久化升级：SQLite → PostgreSQL
□ 会话存储：内存 → Redis，支持多实例
□ 可观测：日志、指标、Trace，记录每次工具调用
□ 限流：防止 API 滥用
□ 工具权限：不同用户可用工具不同
□ 确认机制：删除、修改类操作需二次确认
□ 错误恢复：工具失败时让 LLM 决定重试或降级
□ 上下文压缩：长对话时压缩历史，控制 token 成本
□ 集成：飞书、钉钉、企业微信、Slack
□ 前端：Web UI 或小程序
□ 测试：工具单测、Agent 契约测试、端到端测试

九、常见问题

Q1：没有 API Key 能用吗？ 能。Mock 模式基于关键词路由，覆盖任务、日程、笔记的常见指令，适合学习和演示。

Q2：能用国产模型吗？ 能。任何 OpenAI 兼容接口都行，把 OPENAI_BASE_URL 指向 DeepSeek、通义、Kimi 等即可。

Q3：数据安全吗？ 本地 SQLite，数据不出机器。接入云端 LLM 时，只有对话内容会发送，数据库内容只在工具执行后作为结果返回。

Q4：为什么不用 LangChain？ 本文目的是讲清原理。理解 Agent 循环和工具调用后，用不用框架都是选择。生产项目可以基于本文骨架，再引入 LangGraph 等做复杂编排。

Q5：如何扩展新工具？ 在 tools.py 的 build_registry 里注册一个新 Tool，写好 name、description、parameters 和 func 即可。LLM 会自动看到新工具。

十、结语

WorkBuddy 的价值不在于"用了多强的模型"，而在于把 AI 接入了真实的工作流。从聊天框到能读写的助手，中间隔着的正是本文讲的三个概念：LLM、工具调用、Agent 循环。

你刚刚构建的这个版本，已经具备了一个工作助手的完整骨架：持久化、工具注册、Agent 循环、CLI 和 Web API。接下来往哪个方向走，取决于你的真实需求——接入公司日历、对接项目管理系统、集成 IM 机器人，或者加一个前端。

最重要的是：先跑起来，再迭代。零基础不是障碍，不开始才是。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

WorkBuddy Managed Agents