本文使用的代码仓库：Lsogod/personal-ai-pai，如果对你有帮助欢迎点个 Star。

项目简介：PAI 是一个基于 LangChain / LangGraph 构建的个人 AI 助理，支持智能记账、日程提醒、长期记忆和多端同步（Web + 微信小程序）。

分支说明：

• main — 多节点路由架构（Router → 专业领域节点）
• feat/single-agent — 单 Agent + 丰富工具集架构（本文基于此分支）

很多人第一次接触 Agent 工具调用，都是从 LangChain 的 @tool 开始的。

看起来很简单：写个函数、加个装饰器、把它交给模型，模型就能自己调工具。

但真正把 Agent 跑进业务里，你很快就会发现，@tool 只是入口，远远不是全部。

难点往往不在“怎么声明一个工具”，而在这些更工程化的问题：

• LLM 怎么知道有哪些工具可以用？
• @tool 装饰器到底做了什么？
• LLM 返回的 tool_calls 是什么结构？
• 工具真正执行时，数据库事务、外部 API 调用、运行时状态怎么管理？
• 工具多了以后怎么分组、怎么控制可见范围？

如果这些问题没有设计清楚，最后得到的通常不是“工具系统”，而是一堆能跑、但很快会失控的函数入口。

这篇文章不讲 Demo，而是直接结合一个真实项目里的实现，拆开看一套可落地的 Agent Tool 架构到底长什么样。

你可以把它当成一条从 @tool 到完整执行链的实战路径：

1. LangChain Tool 的基本用法和原理
2. Tool Calling 在模型侧是怎么工作的
3. 工具执行层如何做到安全、可控
4. 不同类型的工具分别怎么设计
5. 如何把工具组织成可扩展的系统
   

---

一、先从最简单的开始：LangChain 的 @tool 装饰器

在 LangChain 里，定义一个工具最简单的方式就是 @tool 装饰器：

from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。"""
    return f"{city}今天晴，25°C"

这三行代码做了什么？

1. 函数名 get_weather 变成了工具名
2. 参数签名 city: str 变成了工具的输入 schema
3. docstring 变成了工具描述，告诉 LLM 这个工具是干什么的

LLM 看到的不是你的 Python 代码，而是一份结构化的工具描述（JSON Schema）。@tool 的本质就是把一个普通函数转换成 LLM 能理解的格式。

但在一个真实项目里，工具不是只写一个函数就结束了。

一套能落地的工具系统，至少要同时解决 4 件事：

• 系统里到底有哪些工具，能力边界怎么定义
• 哪些工具当前对 Agent 可见，哪些不该暴露
• 模型看到的工具接口长什么样，参数和描述怎么设计
• 工具真正执行时，数据库、外部 API、调度器和记忆状态怎么统一收口

换句话说，@tool 只是最外层入口。再往里，还需要能力注册、可见范围控制、执行收口和副作用管理这些工程层，工具系统才能真正跑进业务里。

1. tool_registry.py

这里不负责执行，只负责回答一个问题：

系统里到底有哪些工具。

例如：

def list_builtin_tool_metas() -> list[ToolMeta]:
    return [
        {
            "name": "now_time",
            "source": "builtin",
            "description": "按时区返回当前本地时间。",
            "enabled": True,
        },
        {
            "name": "ledger_text2sql",
            "source": "builtin",
            "description": "通过受保护的 SQL 流程执行自然语言账单增删改查。",
            "enabled": True,
        },
        {
            "name": "memory_save",
            "source": "builtin",
            "description": "将用户明确要求记住的信息直接写入长期记忆。",
            "enabled": True,
        },
    ]

这一层更像能力目录，而不是执行层。

2. toolsets.py

当工具数量增多后，你需要一层来管理”哪些工具对 Agent 可见”。这一层的职责是：

• 定义 Agent 当前可见的完整工具集合
• 按领域划分工具分组（账单类、日程类、记忆类等），方便复用和扩展

例如：

SHARED_TOOL_NAMES: set[str] = {
    "now_time",
    "fetch_url",
}

MCP_TOOL_NAMES: set[str] = {
    "mcp_list_tools",
    "mcp_call_tool",
    "maps_weather",
}

LEDGER_TOOL_NAMES: set[str] = {
    "analyze_receipt",
    "ledger_text2sql",
    "ledger_insert",
    "ledger_update",
    "ledger_delete",
    "ledger_get_latest",
    "ledger_list_recent",
    "ledger_list",
}

最终，把各组工具合并成 Agent 实际可见的完整集合：

MAIN_AGENT_TOOL_NAMES: set[str] = (
    SHARED_TOOL_NAMES
    | VISION_TOOL_NAMES
    | MCP_TOOL_NAMES
    | CONVERSATION_TOOL_NAMES
    | LEDGER_TOOL_NAMES
    | SCHEDULE_TOOL_NAMES
    | PROFILE_TOOL_NAMES
)

这一步非常关键：工具不是”全量裸暴露”给 LLM，而是先按领域分组，再组合进 Agent 的可见边界。这样做的好处是，后续新增或下线某组工具时，只需要改一个集合定义。

3. langchain_tools.py

这一层是真正”暴露给 LLM”的地方。

LLM 看不到数据库模型，也看不到 tool_executor.py 里的内部判断。
LLM 真正看到的是 @tool(...) 包装后的接口。

例如：

from langchain.tools import ToolRuntime, tool

# AgentToolContext 是只包含 user_id / conversation_id / image_urls
# 这类可序列化字段的精简运行时上下文。
@tool("ledger_insert")
async def ledger_insert_tool(
    amount: float,
    category: str,
    item: str,
    transaction_date: str = "",
    image_url: str = "",
    *,
    runtime: ToolRuntime[AgentToolContext],
) -> str:
    """插入一条账单记录，并返回 JSON 行数据。"""
    return await _run_tool(
        runtime=runtime,
        source="builtin",
        name="ledger_insert",
        args={
            "amount": amount,
            "category": category,
            "item": item,
            "transaction_date": transaction_date,
            "image_url": image_url,
        },
)

也就是说，LLM 真正接收到的是工具名、参数签名和工具描述，而不是底层实现细节。这部分在第二节会展开说明。

工具运行时上下文是通过 ToolRuntime 注入的。像 user_id、conversation_id、image_urls 这类可序列化字段，会放进 runtime.context；而 audit_hook 这类 Callable 运行时对象，不会直接进入 context_schema，而是改走 ContextVar 一类的运行时通道。

这里还有一个不容易理解的细节：

• mcp_list_tools / mcp_call_tool 是 LangChain wrapper 暴露给 LLM 的名字
• executor 内部真正执行的 builtin 名称是 tool_list / tool_call
• 这层映射通过 tool_executor.py 里的 BUILTIN_TOOL_ALIAS 完成

也就是说，给模型看的工具名 和 执行层内部的能力名 可以不完全相同，只要映射关系清晰即可。

4. tool_executor.py

这层才是执行收口层。

真正的参数校验、数据库访问、scheduler 操作、记忆状态推进，都是在这里统一处理。

例如：

async def execute_capability(...):
    if src == "builtin":
        if not await is_tool_enabled("builtin", tool_l):
            return _result(False, error=f"tool `{tool_l}` is disabled by admin.")

        if tool_l == "now_time":
            ...

        if tool_l == "ledger_insert":
            ...

        if tool_l == "memory_save":
            ...

这就意味着，工具不是散落在各处执行，而是被统一收口到了一个中心执行入口。

---

二、Tool Calling 的工作原理：LLM 到底看到了什么

理解了 @tool 装饰器之后，下一个问题是：LLM 到底是怎么"调用"工具的？

答案是 Tool Calling（工具调用）。LLM 并不会直接执行你的 Python 函数，而是返回一个结构化的调用请求，告诉你"我想调用哪个工具、传什么参数"，由你的代码去真正执行。

从模型视角来看，一个工具其实只有三样东西：

1. 工具名
2. 参数
3. 描述

例如 schedule_insert：

@tool("schedule_insert")
async def schedule_insert_tool(
    content: str,
    trigger_time: str,
    status: str = "PENDING",
    job_id: str = "",
    *,
    runtime: ToolRuntime[AgentToolContext],
) -> str:
    """创建一条日程提醒，并返回 JSON 行数据。trigger_time 格式：YYYY-MM-DD HH:MM:SS。"""
    return await _run_tool(
        runtime=runtime,
        source="builtin",
        name="schedule_insert",
        args={
            "content": content,
            "trigger_time": trigger_time,
            "status": status,
            "job_id": job_id,
        },
    )

对于 LLM 来说，这个工具的含义就是：

• 名字叫 schedule_insert
• 需要 content 和 trigger_time
• 可选 status 和 job_id
• 描述告诉它这是“创建提醒”的工具

不过从代码看，这里的 docstring 只是 给模型看的接口提示。真正到了 executor 层，trigger_time 不只支持 YYYY-MM-DD HH:MM:SS，也支持 "10秒后"、"5分钟后"、"明天下午3点"、"下周一上午10点" 这类相对或自然语言时间表达。

所以工具设计的第一原则就是：

先把 LLM 该看到的接口设计清楚，再考虑底层实现。

在这个项目里，主 Agent 也是按这个思路消费工具调用能力的。只不过在当前 LangChain 1.x 写法里，不再是手动拼一个 create_react_agent，而是直接用 create_agent(...)，并通过 astream_events(...) 监听工具开始/结束事件：

agent = create_agent(
    model=get_llm(node_name=LLM_NODE_NAME),
    tools=tools,
    system_prompt=system_prompt,
    context_schema=AgentToolContext,
    name=f"main_agent_{user_id}_{conversation_id or 0}",
)

async for event in agent.astream_events(
    {"messages": [{"role": "user", "content": effective_content}]},
    context=ctx,
    config={"recursion_limit": 12},
    version="v2",
):
    ...

底层原理没有变：模型依然会产出结构化的工具调用意图，只是当前代码不是自己手写解析每一步 tool_calls，而是借助 LangChain 1.x 的 Agent Runtime 和事件流来拿到工具开始、工具结束、最终回答这些关键节点。

一个典型的 tool call，从工程上可以近似理解为下面这种结构：

[
  {
    "name": "ledger_insert",
    "args": {
      "amount": 28,
      "category": "餐饮",
      "item": "午饭"
    },
    "id": "call_xxx",
    "type": "tool_call"
  }
]

也就是说，所谓 Tool Calling，本质上不是“模型自动执行函数”，而是“模型返回一条结构化调用意图，运行时再去执行”。

如果 wrapper 层设计混乱，模型就很难正确调用。

这里有一个很关键但很容易被忽略的点：
模型并不是"读懂了这段 Python 代码，然后决定怎么执行"，而是 LangChain 在运行前已经把 @tool(...) 包装过的函数转换成了工具 schema，再把这份 schema 发给模型。

以 ledger_insert 为例，这段代码对模型来说，会被转换成一种更接近下面这样的结构化定义：

{
  "name": "ledger_insert",
  "description": "插入一条账单记录，并返回 JSON 行数据。",
  "parameters": {
    "type": "object",
    "properties": {
      "amount": { "type": "number" },
      "category": { "type": "string" },
      "item": { "type": "string" },
      "transaction_date": { "type": "string" },
      "image_url": { "type": "string" }
    },
    "required": ["amount", "category", "item"]
  }
}

也就是说，模型真正依赖的是三类信息：

• 工具名：决定这是不是自己该调用的能力
• 参数名和参数类型：决定应该填什么字段
• docstring 描述：决定这个工具到底是干什么的

所以当用户说"记一笔午饭 28 元"时，模型会去匹配：

• 有没有一个叫 ledger_insert 的工具
• 这个工具需不需要 amount/category/item
• 它的描述是不是和"插入一条账单记录"一致

如果这些信息设计得足够清楚，模型就比较容易产出正确的 tool call。

所以从工程角度看，一个工具是否好用，不只是底层逻辑写得对不对，还取决于：

• 工具名是否清晰
• 参数命名是否贴近用户语言
• docstring 是否能准确描述用途

---

三、工具执行层：如何让工具安全、可控地操作数据库和外部系统

前面讲的都是”LLM 怎么看到工具”，但工具系统真正的难点不在接口设计，而在执行层——工具被调用后，真正的副作用怎么控制。

一个工具可能要写数据库、调外部 API、注册定时任务。如果这些操作散落在各处，系统很快就会失控。

1. 数据库 session 统一来自 runtime context

普通工具执行时，数据库 session 不是随便 new 的，而是统一通过 ContextVar 注入：

def get_session() -> AsyncSession:
    session = _session_ctx.get()
    if session is None:
        raise RuntimeError("session context not set")
    return session

这意味着：

• 一个请求链路先把 session 放进上下文
• executor 再从上下文里取出 session
• 同一条工具执行链保持统一上下文

当然也有例外：

• ledger_text2sql.py 这种复杂模块会自己开独立 AsyncSessionLocal()
• Profile 内联工具（update_user_profile / query_user_profile）也会自己开 AsyncSessionLocal()

但普通工具的主路径都是 runtime context。

补一句实现层细节：这个“runtime context”在当前项目里已经拆成两类。

• AgentToolContext：给 agent / tool schema 暴露的可序列化字段
• ContextVar：承载 session、scheduler、sender、audit hook 这类运行时对象

2. 用户越权靠 user_id 显式收口

几乎所有操作用户数据的工具，都会把 user_id 作为第一层过滤条件。

例如账单查询：

stmt = select(Ledger).where(Ledger.user_id == uid)

例如更新账单：

ledger = await session.get(Ledger, ledger_id)
if not ledger or ledger.user_id != user_id:
    return None

例如查找记忆：

target = await find_active_long_term_memory(
    session=session,
    user_id=uid,
    memory_id=memory_id,
    memory_key=memory_key,
    content_hint=target_hint,
    memory_type=memory_type,
)

所以这个项目里最重要的执行原则之一是：

工具的任何数据库访问都不能脱离 user_id。

3. commit / rollback 是显式控制的

简单 CRUD 工具一般在业务函数里显式 commit()：

session.add(ledger)
await session.commit()
await session.refresh(ledger)

复杂工具遇到失败会显式 rollback()，例如 ledger_text2sql：

try:
    result = await db.execute(stmt, params)
except Exception:
    await db.rollback()

这就把事务边界说清楚了：

• 普通工具：本地小事务，直接提交
• 复杂工具：每一阶段都明确 commit / rollback

4. scheduler 也是统一上下文对象

日程工具除了数据库，还会访问 scheduler：

def get_scheduler() -> SchedulerService:
    scheduler = _scheduler_ctx.get()
    if scheduler is None:
        raise RuntimeError("scheduler context not set")
    return scheduler

所以 schedule_insert/update/delete 不只是数据库操作，还会：

• add_job
• remove_job

5. memory 工具还要推进消息状态

这类工具最容易被低估，因为它不是只写长期记忆表。

例如：

async def _mark_source_message_memory_processed(...):
    row.memory_status = "PROCESSED"
    row.memory_processed_at = datetime.now(ZoneInfo("UTC"))
    row.memory_error = None
    ...
    await session.commit()

这说明 memory_save/memory_append/memory_delete 的副作用至少有两层：

• 改 long_term_memories
• 改 messages / conversations 的状态

这就是为什么记忆工具不是普通 CRUD，而是状态型工具。

---

四、工具的 4 种实现模式

当工具数量增多后，你会发现不同工具的复杂度差异很大。有的只返回一个字符串，有的要操作数据库还要同步调度器。

从实现角度看，工具大致可以分成四类：

1. 纯轻量工具

特点：

• 不查库
• 没有复杂副作用
• executor 几行就能写完

代表：

• now_time

2. 标准 CRUD 工具

特点：

• wrapper 很薄
• executor 负责参数校验和编排
• 底层函数负责真正数据库事务

代表：

• ledger_insert
• ledger_update
• ledger_delete
• ledger_list
• schedule_insert
• schedule_update
• schedule_delete
• schedule_list

3. 复杂规划工具

特点：

• 不能只写 executor 分支
• 必须拆独立模块
• LLM 负责规划，系统负责安全边界

代表：

• ledger_text2sql
• analyze_receipt
• analyze_image

4. 状态型工具

特点：

• 不只是改业务数据
• 还会推进系统状态

代表：

• memory_save
• memory_append
• memory_delete
• conversation_current
• conversation_list
• update_user_profile

这四类基本覆盖了当前项目里绝大多数工具的实现方式。

---

五、4 类工具的代码实现详解

下面选 4 个最有代表性的工具，分别展示每种类型的实现方式。

1. now_time：最简单的轻量工具

这是整个系统里最标准的最小工具。

wrapper

@tool("now_time")
async def now_time_tool(
    timezone: str = "Asia/Shanghai",
    *,
    runtime: ToolRuntime[AgentToolContext],
) -> str:
    """按时区名称返回当前本地时间，例如：Asia/Shanghai。"""
    return await _run_tool(
        runtime=runtime,
        source="builtin",
        name="now_time",
        args={"timezone": timezone},
    )

executor

if tool_l == "now_time":
    timezone = str(params.get("timezone") or settings.timezone or "Asia/Shanghai").strip()
    return _result(True, output=_render_now_time(timezone))

实现要点

这个工具体现了最小实现路径：

• wrapper 极薄
• executor 极薄
• 没有数据库
• 没有独立模块

这类工具的实现重点不在业务逻辑，而在于保持 wrapper 和 executor 都足够薄。

---

2. ledger_insert：标准 CRUD 工具

这是最值得学习的普通业务工具。

wrapper

@tool("ledger_insert")
async def ledger_insert_tool(
    amount: float,
    category: str,
    item: str,
    transaction_date: str = "",
    image_url: str = "",
    *,
    runtime: ToolRuntime[AgentToolContext],
) -> str:
    """插入一条账单记录，并返回 JSON 行数据。"""
    return await _run_tool(
        runtime=runtime,
        source="builtin",
        name="ledger_insert",
        args={
            "amount": amount,
            "category": category,
            "item": item,
            "transaction_date": transaction_date,
            "image_url": image_url,
        },
    )

executor

if tool_l == "ledger_insert":
    uid = _resolve_user_id(params.get("user_id", user_id))
    if uid <= 0:
        return _result(False, error="missing required arg: user_id")

    amount_raw = params.get("amount")
    try:
        amount = float(amount_raw)
    except Exception:
        amount = 0.0
    if amount <= 0:
        return _result(False, error="invalid amount")

    category = str(params.get("category") or "其他").strip() or "其他"
    item = str(params.get("item") or "消费").strip() or "消费"
    transaction_date = _parse_utc_naive_arg(params.get("transaction_date")) or datetime.utcnow()
    image_url = str(params.get("image_url") or "").strip() or None

    row = await insert_ledger(
        session=get_session(),
        user_id=uid,
        amount=amount,
        category=category,
        item=item,
        transaction_date=transaction_date,
        image_url=image_url,
        platform=platform,
    )
    payload = _ledger_to_payload(row)
    return _result(True, output=json.dumps(payload, ensure_ascii=False), output_data=payload)

底层数据库函数

async def insert_ledger(
    session: AsyncSession,
    user_id: int,
    amount: float,
    category: str,
    item: str,
    transaction_date: Optional[datetime] = None,
    image_url: Optional[str] = None,
    platform: str = "",
) -> Ledger:
    ledger = Ledger(
        user_id=user_id,
        amount=amount,
        category=category,
        item=item,
        transaction_date=transaction_date or datetime.utcnow(),
        image_url=image_url,
    )
    session.add(ledger)
    await session.commit()
    await session.refresh(ledger)
    return ledger

实现要点

ledger_insert 这类工具的实现重点在于：

• wrapper 不做业务校验
• executor 负责参数检查和时间归一化
• 真正写库放到独立业务函数里

这也是当前项目里标准 CRUD 工具的典型写法。

---

3. ledger_text2sql：复杂规划工具

这是当前工具系统里最复杂也最有代表性的工具。

如果它写得不好，系统就会非常危险；
写得好，它就是一个非常好的工程范例。

为什么它不能写成普通 CRUD

因为它面对的输入不是结构化字段，而是自然语言。

所以正确做法不是：

• 模型生成 SQL
• 系统直接执行

而是：

1. 让模型先生成结构化计划
2. 系统再审计 SQL
3. 系统决定是直接执行还是 preview / commit

结构化计划

class LedgerText2SQLPlan(BaseModel):
    matched: bool = Field(default=False)
    intent: str = Field(default="unknown")
    sql: str = Field(default="")
    params: dict[str, Any] = Field(default_factory=dict)
    summary: str = Field(default="")
    confidence: float = Field(default=0.0)

规划层

async def _plan_sql(message: str, conversation_context: str = "") -> dict[str, Any]:
    llm = get_llm(node_name="ledger_text2sql")
    runnable = llm.with_structured_output(LedgerText2SQLPlan)
    system_prompt = (
        "你是 PostgreSQL 的账单 Text-to-SQL 规划器。\n"
        "只能操作 ledgers 表。\n"
        "必须且只能返回一个 json 对象。\n"
        "只返回结构化字段：matched, intent, sql, params, summary, confidence。\n"
        "intent 只能是 select/insert/update/delete/unknown 之一。\n"
        "对于 select/update/delete：SQL 必须包含 WHERE user_id = :user_id。\n"
        "对于 insert：插入列中必须显式包含 user_id。\n"
        "ledgers.transaction_date 在数据库中按 UTC-naive 存储。\n"
        "相对时间表达必须先按用户本地时区计算，再换算为 UTC-naive 参数。\n"
    )
    result = await runnable.ainvoke(...)

安全审计层

def _is_safe_sql(sql: str, intent: str, user_message: str) -> tuple[bool, str]:
    stmt = _strip_single_statement(sql)
    lower = stmt.lower()
    if not stmt:
        return False, "empty_sql"
    if ";" in stmt:
        return False, "multi_statement_not_allowed"
    if _FORBIDDEN_SQL.search(stmt):
        return False, "forbidden_keyword"
    if _OTHER_TABLES.search(stmt):
        return False, "non_ledger_table_detected"
    if " ledgers" not in f" {lower} ":
        return False, "must_target_ledgers"

直接执行层

async def try_execute_ledger_text2sql(
    user_id: int,
    message: str,
    conversation_context: str = "",
) -> str | None:
    plan = await _plan_sql(message, conversation_context)
    if not plan or not plan.get("matched"):
        return None

    intent = str(plan.get("intent") or "unknown").strip().lower()
    confidence = float(plan.get("confidence") or 0.0)
    if confidence < 0.60:
        return None

    sql = _strip_single_statement(str(plan.get("sql") or ""))
    ok, reason = _is_safe_sql(sql, intent, message)
    if not ok:
        return f"该账单操作被安全策略拦截：{reason}。请换一种更明确的说法。"

实现要点

这个工具真正有价值的地方不在 SQL 本身，而在于：

• 复杂工具不能把所有逻辑塞进 executor
• LLM 适合做规划，不适合做最终边界控制
• 真正危险的操作必须用系统规则兜底

---

4. memory_save：状态型工具

这类工具和普通 CRUD 最大的区别是：

它们不只是改主表，还要推进状态。

executor

if tool_l == "memory_save":
    processed = await upsert_long_term_memories(
        session=session,
        user_id=uid,
        conversation_id=conversation_id,
        source_message_id=(source_message_id or None),
        candidates=[
            {
                "op": "save",
                "memory_type": memory_type,
                "key": memory_key,
                "content": content,
                "importance": importance,
                "confidence": confidence,
                "ttl_days": ttl_days,
            }
        ],
        user_text=f"用户明确要求记住：{content}",
        bypass_refine=True,
    )
    await _mark_source_message_memory_processed(
        session=session,
        user_id=uid,
        conversation_id=conversation_id,
        source_message_id=(source_message_id or None),
    )

状态推进

async def _mark_source_message_memory_processed(...):
    row.memory_status = "PROCESSED"
    row.memory_processed_at = datetime.now(ZoneInfo("UTC"))
    row.memory_error = None
    session.add(row)
    ...
    await session.commit()

实现要点

这个工具最关键的点在于：

• 有些工具改的不只是业务数据
• 还要改 message / conversation 的流程状态
• 这类工具必须把“状态推进”看成自己的一部分

---

六、实战：如何新增一个工具

理解了工具的分类和执行流程后，新增一个工具其实就是按模式套用。下面是四种常见场景：

情况 1：简单工具

如果它：

• 不查库
• 没副作用
• 纯返回文本或简单结果

典型做法是：

1. 在 tool_registry.py 注册
2. 在 toolsets.py 放入对应分组，并确保它进入 MAIN_AGENT_TOOL_NAMES
3. 在 langchain_tools.py 写 @tool(...) wrapper，并按需接入 ToolRuntime
4. 在 tool_executor.py 写一个简单分支

情况 2：标准 CRUD 工具

如果它：

• 有数据库访问
• 逻辑不复杂
• 参数结构明确

典型做法是：

1. wrapper 只负责暴露参数
2. executor 做参数校验
3. 底层业务函数做事务提交

情况 3：复杂规划工具

如果它：

• 输入是自然语言
• 有多阶段执行
• 有安全风险

典型做法是：

1. wrapper 尽量薄
2. executor 只做 mode 分发
3. 真正算法拆独立模块
4. 系统控制最终执行边界

情况 4：状态型工具

如果它：

• 不只是改主数据
• 还会影响异步流程或消息状态

典型做法是：

1. 改主表
2. 同时推进状态表
3. 明确幂等和失败后的重试语义

---

七、完整工具系统解决了什么问题

回到最开始的问题：为什么不能只写几个 @tool 函数？因为一个完整的工具系统需要把 LLM 的函数调用变成 一条可控的执行链：

• wrapper 负责让模型容易调用
• executor 负责让系统稳定执行
• 复杂逻辑拆模块
• 状态类副作用显式收口

这也是为什么这个项目里的工具不只是”能用”，而是”可以持续扩展”的原因。

从具体业务来看，每个领域的工具不是一个”万能函数”，而是形成了从快捷路径到开放路径的层次。以账单为例，同时存在三条能力链：

• 快路径：ledger_get_latest / ledger_list_recent——高频定位，降低模型参数构造成本
• 稳定路径：ledger_insert / ledger_update / ledger_delete / ledger_list——结构化 CRUD，确定性强
• 智能路径：analyze_receipt / ledger_text2sql——视觉输入和自然语言查询，覆盖复杂场景

日程工具也类似，但多了一层关键差异：每次 insert/update/delete 都要同步维护运行时 scheduler 的 job 状态，不只是改数据库。

这种分层设计让主 Agent 可以根据问题结构主动切换策略，而不是只有一种解法。

---

八、附录：完整工具速查表

下表列出本文配套项目中的全部工具，按分组和实现类型归类，可以作为设计自己工具系统时的参考：

分组	工具	类型	一句话说明
Shared	now_time	轻量	返回指定时区当前时间
Shared	fetch_url	轻量	通过 MCP fetch 抓取网页内容
Vision	analyze_image	复杂规划	通用视觉问答，结构化输出
Vision	analyze_receipt	复杂规划	小票/截图识别，提取记账字段
MCP	mcp_list_tools	轻量	列出当前可用的外部 MCP 工具（executor 内部别名为 tool_list）
MCP	mcp_call_tool	轻量	通用 MCP 工具转发，带 allowlist 保护（executor 内部别名为 tool_call）
MCP	maps_weather	轻量	天气查询的 MCP 友好别名
Ledger	ledger_insert	标准 CRUD	插入一条账单记录
Ledger	ledger_update	标准 CRUD	局部更新账单字段
Ledger	ledger_delete	标准 CRUD	按 id 删除账单
Ledger	ledger_get_latest	标准 CRUD	快捷查最新一条账单
Ledger	ledger_list_recent	标准 CRUD	快捷查最近 N 条账单
Ledger	ledger_list	标准 CRUD	按时间/分类/关键词条件查询
Ledger	ledger_text2sql	复杂规划	自然语言→SQL 规划→安全审计→执行
Schedule	schedule_insert	标准 CRUD + 运行时	创建提醒并注册 scheduler job
Schedule	schedule_update	标准 CRUD + 运行时	更新提醒并重建 scheduler job
Schedule	schedule_delete	标准 CRUD + 运行时	删除提醒、delivery 记录和 job
Schedule	schedule_get_latest	标准 CRUD	快捷查最新一条提醒
Schedule	schedule_list_recent	标准 CRUD	快捷查最近 N 条提醒
Schedule	schedule_list	标准 CRUD	按时间范围/状态条件查询
Conversation	conversation_current	状态型	获取或创建当前激活会话
Conversation	conversation_list	状态型	列出会话并标注 active 状态
Memory	memory_list	状态型	列出长期记忆，过滤过期/identity，更新访问时间
Memory	memory_save	状态型	写入长期记忆并推进消息状态
Memory	memory_append	状态型	定位目标记忆并增量追加内容
Memory	memory_delete	状态型	定位目标记忆并删除，推进消息状态
Profile	update_user_profile	状态型（内联）	更新昵称/AI名称/表情，清理 identity 记忆
Profile	query_user_profile	轻量（内联）	返回用户档案文本

说明：Profile 工具是当前系统里的一个特例——update_user_profile 和 query_user_profile 都没有走 tool_executor.py，而是直接在 langchain_tools.py 里内联实现。前者需要在更新后立刻清理 identity 类长期记忆，后者则直接读取用户档案并返回文本。

---

总结

如果只看 LangChain 的表层 API，很容易以为工具系统就是“写几个 @tool 函数，交给模型去调”。

但真正可落地的 Agent Tool 架构，至少还需要再往下多做几层：

1. @tool 装饰器是起点——它把 Python 函数转换成 LLM 能理解的工具描述
2. Tool Calling是机制——LLM 不执行代码，而是返回结构化的调用请求
3. 执行层是关键——真正的数据库操作、API 调用、状态管理都需要统一收口
4. 分层设计是方法——注册、分组、包装、执行各司其职，工具才能可维护、可扩展

换句话说，@tool 解决的是“模型怎么看见工具”，而 ToolRuntime、create_agent(...)、tool_executor、运行时上下文、事务边界、调度器、副作用状态推进，解决的才是“工具怎么安全地活在生产环境里”。

如果你正在搭建自己的 Agent 工具系统，最值得优先做好的，通常是这几件事：

• 先用 @tool 快速跑通一个最简单的工具调用链路
• 工具超过 5 个时，开始考虑分组和可见范围控制
• 涉及数据库写操作时，一定要有统一的执行入口，不要让工具直接散写 SQL
• 复杂业务逻辑拆成独立模块，工具只做薄薄一层转发

这也是这套实现里最想说明的一点：

好的工具系统，不是让模型“看起来很聪明”，而是让整个执行链在复杂业务下依然可控。

GitHub 仓库：Lsogod/personal-ai-pai

如果你对这种“单 Agent + 丰富工具集”或者“多节点路由架构”的实现路线感兴趣，欢迎点个 Star。

参考源码

• tool_registry.py
• toolsets.py
• langchain_tools.py
• tool_executor.py
• finance.py
• vision.py
• ledger_text2sql.py
• memory.py