好的,以下是根据我们讨论所达成的共识,针对 valid_sql
校验流程与 analyze_validation_error
路由逻辑的最终建议报告。
valid_sql(sql: str)
工具函数增强(在 sql_tools.py
中) vn.run_sql(sql + ' LIMIT 0')
try/except
捕获字段或表不存在等运行时错误。@tool
def valid_sql(sql: str) -> str:
...
try:
vn.run_sql(sql + " LIMIT 0")
except Exception as e:
return f"SQL验证失败:执行失败。详细错误:{str(e)}"
return "SQL验证通过:语法正确且字段存在"
_async_update_state_after_tool_node
方法保持不变(在 agent.py
中)elif tool_name == 'valid_sql':
if "失败" in tool_output:
next_step = 'analyze_validation_error'
else:
next_step = 'run_sql'
analyze_validation_error
不是工具也不是节点,仅是对 LLM 的策略建议;suggested_next_step
完成。_async_agent_node
中针对 analyze_validation_error
提供 LLM 指导(重点)state['suggested_next_step'] == 'analyze_validation_error'
valid_sql
说明:上一步 SQL 验证失败。
- 如果是语法错误,请尝试修复语法错误,并调用 valid_sql 工具重新验证 SQL 是否有效;
- 如果是字段或表名不存在等问题,请告诉用户缺少的字段或表名,并直接向用户返回基于常识的解释或答案。
_async_agent_node
):next_step = state.get("suggested_next_step")
if next_step and next_step != "analyze_validation_error":
instruction = f"Suggestion: Consider using the '{next_step}' tool for the next step."
messages_for_llm.append(SystemMessage(content=instruction))
if next_step == "analyze_validation_error":
for msg in reversed(state["messages"]):
if isinstance(msg, ToolMessage) and msg.name == "valid_sql":
messages_for_llm.append(SystemMessage(content=(
"说明:上一步 SQL 验证失败。\n"
"- 如果是语法错误,请尝试修复语法错误,并调用 valid_sql 工具重新验证 SQL 是否有效;\n"
"- 如果是字段或表名不存在等问题,请告诉用户缺少的字段或表名,并直接向用户返回基于常识的解释或答案。"
)))
break
模块 | 状态 | 操作建议 |
---|---|---|
valid_sql 工具 |
✅ 增强完成 | 添加 run_sql(... LIMIT 0) 检查字段 |
update_state_after_tool |
✅ 保持不变 | 继续使用 'analyze_validation_error' |
_async_agent_node |
✅ 需要优化 | 区分是否为 analyze 分支,添加具体指导语句 |