LangChain SQL Agent非精确匹配返回随机ID问题求助

开发了一个SQL Agent，用于查询包含id和email字段的数据库表，需求为：匹配到精确email（忽略大小写）则返回对应id，无匹配则返回NONE。精确匹配时功能正常，但输入近似email（无精确匹配）时，Agent会返回id列的随机值，例如输入myFirstName.LastnamA@mydomain.com时出现该问题。

核心问题是LLM生成的SQL逻辑存在漏洞：当无精确匹配时，Agent可能生成了未加条件的查询（如SELECT id FROM users），或错误处理了空结果，导致返回随机id。即使prompt明确要求无匹配返回NONE，LLM仍可能未正确处理查询结果为空的场景，反而直接返回表中任意id。

1. 强制SQL返回明确的空结果标识 #

修改prompt，要求LLM生成的SQL直接在查询中处理空结果，避免Agent自行判断。可以用SQL的COALESCE或CASE语句实现：

• 用COALESCE的SQL示例：

SELECT COALESCE((SELECT id FROM users WHERE LOWER(email) = LOWER('myFirstName.LastnamA@mydomain.com')), 'NONE') AS result;

• 用CASE的SQL示例：

SELECT CASE 
         WHEN EXISTS(SELECT 1 FROM users WHERE LOWER(email) = LOWER('myFirstName.LastnamA@mydomain.com')) 
         THEN (SELECT id FROM users WHERE LOWER(email) = LOWER('myFirstName.LastnamA@mydomain.com')) 
         ELSE 'NONE' 
       END AS result;

2. 严格约束prompt的SQL生成规则 #

在prompt中明确限定生成逻辑，消除LLM的歧义空间：

agent_executor.run("""
执行以下邮箱查询：
1. 必须使用`LOWER(email) = LOWER('目标邮箱')`的条件做精确匹配（忽略大小写）
2. 仅当查询结果有且仅有一条匹配时，返回对应的id
3. 如果查询结果为空，直接返回字符串'NONE'，禁止执行任何无条件的id查询
目标邮箱：myFirstName.LastnamA@mydomain.com
""")

3. 自定义查询工具替代默认SQL工具 #

如果LLM仍无法稳定生成正确SQL，直接编写Python函数实现查询逻辑，并作为自定义工具加入Agent：

from langchain.tools import Tool
from sqlalchemy import create_engine

def get_id_by_target_email(email):
    engine = create_engine("sqlite:///data/users.db")
    with engine.connect() as conn:
        query_result = conn.execute(
            "SELECT id FROM users WHERE LOWER(email) = LOWER(:input_email)",
            {"input_email": email}
        ).fetchone()
        return query_result[0] if query_result else "NONE"

# 定义自定义工具
email_query_tool = Tool(
    name="get_id_by_exact_email",
    func=get_id_by_target_email,
    description="根据精确匹配的邮箱（忽略大小写）获取对应id，无匹配时返回'NONE'"
)

# 创建Agent时加入自定义工具
agent_executor = create_sql_agent(
    llm=OpenAI(temperature=0),
    toolkit=toolkit,
    tools=[email_query_tool],
    verbose=True,
    agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
)

4. 添加结果验证步骤 #

在Agent返回结果前，额外验证返回的id是否对应输入的邮箱：

# 执行Agent查询
result = agent_executor.run("What is the id for email is exactly myFirstName.LastnamA@mydomain.com...")
# 验证结果有效性
if result != "NONE":
    engine = create_engine("sqlite:///data/users.db")
    with engine.connect() as conn:
        verify_result = conn.execute(
            "SELECT 1 FROM users WHERE id = :result_id AND LOWER(email) = LOWER(:target_email)",
            {"result_id": result, "target_email": "myFirstName.LastnamA@mydomain.com"}
        ).fetchone()
        if not verify_result:
            result = "NONE"
print(result)

内容的提问来源于stack exchange，提问作者ML Ops