一文带你了解 Agent 的 Tool 机制：基于 LangChain 从入门到搭建完整工具系统

Lsogod

590人浏览 · 2026-03-23 01:07:50

Lsogod · 2026-03-23 01:07:50 发布

本文使用的代码仓库：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 到完整执行链的实战路径：

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

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

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

from langchain.tools import tool

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

这三行代码做了什么？

函数名 get_weather 变成了工具名
参数签名 city: str 变成了工具的输入 schema
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 函数，而是返回一个结构化的调用请求，告诉你"我想调用哪个工具、传什么参数"，由你的代码去真正执行。

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

工具名
参数
描述

例如 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
系统直接执行

而是：

让模型先生成结构化计划
系统再审计 SQL
系统决定是直接执行还是 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：简单工具

如果它：

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

典型做法是：

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

情况 2：标准 CRUD 工具

如果它：

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

典型做法是：

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

情况 3：复杂规划工具

如果它：

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

典型做法是：

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

情况 4：状态型工具

如果它：

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

典型做法是：

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

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

回到最开始的问题：为什么不能只写几个 @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 类长期记忆，后者则直接读取用户档案并返回文本。