Langchain tool 为什么无法向数据库添加记录？

2026-11-04排障赵师傅

问题中可以确认两处错误：cursor.execute() 的参数传法不正确，写入在提交前就失败了；工具内部的裸 except 又吞掉了异常。修正 SQL 后仍然反复选题、重复提问，则与每轮调用只传入最新一条用户消息有关。

system prompt 中的“允许执行 DML”是给模型的行为指令。要写入数据库，模型需要实际调用工具，再由工具中的 Python 代码执行并提交。增加权限说明或调整 temperature，都无法修复这些代码问题。

为什么停在 `topic executing...`

原来的调用是：

cursor.execute(sqlinsert, None, topic, 0)

sqlite3.Cursor.execute() 接收 SQL 语句和一个参数序列。三个占位符对应的值应一起放进元组：

cursor.execute(sqlinsert, (None, topic, 0))

原写法在调用 Python 方法时就会触发 TypeError，因此不会执行到后面的 topic committing...。

也可以省略自动生成的主键，明确指定要插入的列：

cursor.execute(
    "INSERT INTO TOPICS (Topic, Understood) VALUES (?, ?)",
    (topic, 0),
)

TopicID 是 INTEGER PRIMARY KEY AUTOINCREMENT，插入时可以省略。显式指定列名，也能避免代码依赖表中所有列的排列顺序。

为什么外层 `try except` 没有显示异常

工具内部已经捕获了异常：

except:
    return "Error adding topic"

异常在这里被转换成普通字符串返回，因此外层的以下代码捕获不到它：

except Exception as e:
    print("Error ", e)

模型收到的也只有一句模糊的 "Error adding topic"，无法判断是参数错误、数据库被锁定，还是连接到了错误的文件。

调试时，可以先打印完整 traceback，再重新抛出异常：

except Exception:
    import traceback

    traceback.print_exc()
    raise

这会输出完整 traceback。重新抛出的异常如何继续传播，取决于 agent 的工具异常处理配置。

如果希望工具在失败后仍向模型返回结果，应在结果中保留明确的错误类型和原因，并在程序日志中记录 traceback。

先验证工具能够独立写入

接入 agent 前，先直接调用工具，区分数据库代码出错和模型没有调用工具这两种情况。

下面沿用项目已有的 @tool 导入，假设 TOPICS 表已经创建：

import logging
import sqlite3
from contextlib import closing
from pathlib import Path

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 初始化数据库、工具写入和查询验证，都应使用同一个路径。
DB_PATH = Path("learn.db").resolve()
print("Database path:", DB_PATH)


@tool
def add_topic(topic: str) -> dict:
    """添加一个具体的 Python 学习主题，返回写入结果和 TopicID。"""
    topic = topic.strip()

    if not topic:
        return {
            "ok": False,
            "error": "topic must not be empty",
        }

    try:
        with closing(sqlite3.connect(str(DB_PATH))) as connection:
            with connection:
                cursor = connection.execute(
                    "INSERT INTO TOPICS (Topic, Understood) VALUES (?, ?)",
                    (topic, 0),
                )
                topic_id = cursor.lastrowid

        return {
            "ok": True,
            "topic_id": topic_id,
            "topic": topic,
        }

    except sqlite3.Error as exc:
        logger.exception("Failed to add topic")
        return {
            "ok": False,
            "error_type": type(exc).__name__,
            "error": str(exc),
        }

with connection: 在正常退出时提交事务，发生异常时回滚；外层的 closing(...) 负责关闭连接。单独使用 SQLite 连接的上下文管理器，不会自动关闭连接。

直接调用工具，再查询数据库确认结果：

result = add_topic.invoke({"topic": "Python function parameters"})
print(result)

with closing(sqlite3.connect(str(DB_PATH))) as connection:
    rows = connection.execute(
        "SELECT TopicID, Topic, Understood FROM TOPICS"
    ).fetchall()
    print(rows)

这段操作会实际添加一条记录，适合在测试数据库中执行。直接调用失败，就先根据日志修复数据库问题；调用成功后，再检查 agent 的工具调用过程。

还要核对数据库路径。sqlite3.connect("learn.db") 使用相对于当前工作目录的路径，运行目录变化后，可能创建或打开另一个同名文件。数据库查看器也必须打开打印出的那个文件。

数据库初始化完成后，应显式提交并关闭连接，避免连接长期占用资源。

修正多轮调用，保留上一轮上下文

原来的循环每次只传入最新输入：

agent.invoke({
    "messages": [
        {"role": "user", "content": input(">>> ")}
    ]
})

如果没有额外配置会话持久化，下一次调用就不知道上一轮问了什么，也不知道用户正在回答哪道题。例如，模型单独收到“它们存储信息”，很难判断这句话对应什么题目。

SQLite 中的记录不会自动进入模型上下文。程序需要传递会话历史，或通过工具读取相关状态。

沿用题目中的 create_agent 调用方式，可以先传入完整消息历史来验证：

agent = create_agent(
    model=model,
    tools=tools,
    system_prompt=system_prompt,
)

messages = []

while True:
    try:
        user_input = input(">>> ")
    except (EOFError, KeyboardInterrupt):
        break

    if user_input.strip().lower() in {"exit", "quit"}:
        break

    messages.append({
        "role": "user",
        "content": user_input,
    })

    result = agent.invoke({"messages": messages})
    messages = result["messages"]

    print(messages[-1].content, flush=True)

这里保留了完整的 messages，包括模型的工具调用和工具返回结果。只保存最后一条自然语言回复，会丢失工具执行信息。

这种方式只在当前进程中保存历史。如果需要重启后继续会话，可以根据实际安装的 LangChain/LangGraph 版本配置持久化，或由应用保存并恢复状态。

检查模型是否真的调用了工具

模型说“我会调用 add_question”或“主题已经存在”，不能证明工具执行过。

调试时，可以在某次 agent.invoke() 后检查消息：

for message in result["messages"]:
    tool_calls = getattr(message, "tool_calls", None)
    if tool_calls:
        print("Tool calls:", tool_calls)

    if getattr(message, "type", None) == "tool":
        print(
            "Tool result:",
            getattr(message, "name", None),
            message.content,
        )

检查以下几点：

是否实际调用了 add_topic，参数是否正确。
工具返回的是成功结果还是错误信息。
写入是否已提交，查询的是否为同一个数据库文件。

修正 SQL 后的新日志中，没有出现原先的工具执行打印，因此需要检查工具是否被调用。仅凭这些 CLI 文本，还不能确定原因。

原工具没有查询重复主题的逻辑，表结构中的 Topic 也没有 UNIQUE 约束，因此这段实现无法支持“主题已经存在”的判断。数据库状态应以查询结果和工具返回值为准。

怎样调整提示词和交互流程

提示词可以明确约定工具调用顺序，例如：

确定一个具体的 Python 学习主题后，先调用 add_topic。

只有 add_topic 返回 ok=true 后，才继续保存该主题的问题。
使用工具返回的 topic_id 关联问题，不得自行编造 ID。

数据库是否写入成功、记录是否存在，必须以工具结果为依据。
工具返回 ok=false 时，说明具体错误，不得声称操作成功。

每轮只向用户提出一道题。
收到回答后，保存回答、反馈和尝试次数，再继续当前主题。
当前主题的五道题全部回答正确后，才进入下一个主题。

原提示词中的 OVERRIDEN 无法清楚表达“全部回答正确”，应直接写明完成条件。

工具也要支持这些操作。原问题没有给出 gen_questions 和 add_question 的实现，无法确认它们能否更新回答、追加反馈或增加 Attempts。如果工具只支持插入，提示词要求的查询和更新就无法执行。

对于“必须先保存五道题，再逐题提问”这样的固定流程，由应用代码控制步骤更可靠：

让模型生成主题和五道题。
应用校验结果，并在事务中保存主题及问题。
保存成功后展示第一题。
应用记录当前 TopicID、QID 和完成情况；每轮保存回答与反馈后再推进。

模型负责出题和评价，程序执行数据库事务并推进流程。这样，即使模型漏掉一次工具调用，也不会直接跳过必要的写入步骤。

其他注意事项

在 docstring 中写“主题必须唯一”，不会形成数据库约束。要防止重复写入，应先定义主题名称的规范化规则，再通过 UNIQUE 约束或唯一索引落实；已有的重复数据需要先处理。

“禁止删除”也应通过工具能力落实。只暴露必要的查询、插入和更新操作，避免提供可执行任意 SQL 的入口。

题目使用远程 ChatGroq 推理，PyTorch was not found 警告无法解释这次 sqlite3 参数调用失败。排查这处错误，无需先安装 PyTorch。

可以继续使用自定义 sqlite3 工具，无须改用 langchain_community 的 SQLDatabase tools。先确认工具能够独立写入，再补齐会话状态和流程控制，之后才能判断剩余问题是否与模型的工具调用表现有关。