这是本人在前段时间初次接触多智能体技术的一次实践，增长了许多见识，在此记录一下。

任务需求

这是2025年腾讯游戏算法大赛数智决策科学赛道的题目，具体需求为通过用户提出的需求（自然语言文字），自动生成SQL语句实现用户的需求，实现Text to SQL。
例如：

##输入问题格式为：
{
    "sql_id": "sql_3",
    "question": "统计2025年1月勇者盟约端游活跃玩家交叉峡谷端游及手游活跃玩家\n输出：玩家gplayerid",
    "复杂度": "简单",
    "table_list": [
        "dws_argothek_oss_login_di",
        "dim_argothek_gplayerid2qqwxid_df",
        "dws_mgamejp_login_user_activity_di"
      ],
    "knowledge": "峡谷筛选逻辑:\nsgamecode = \"initiatived\" -- 筛选峡谷游戏\nand saccounttype = \"-100\" -- 账号体系，取-100表示汇总\nand suseridtype in (\"qq\", \"wxid\") -- 账号类型，取qq或wxid\nand splattype in (\"-100\", \"PC\") -- 峡谷手游玩家及PC端玩家\nand splat = \"-100\" -- 写死为-100"
    },

##得到输出结果
{
"sql_id": "sql_3",
"sql": "SELECT DISTINCT df.vgameappid AS gplayerid\nFROM dim_argothek_gplayerid2qqwxid_df df\nJOIN dws_argothek_oss_login_di oss ON df.iuserid = oss.iuserid\nJOIN dws_mgamejp_login_user_activity_di mgame ON df.suserid = mgame.suserid\nWHERE \n oss.statis_date BETWEEN 20250101 AND 20250131\n AND mgame.dtstatdate BETWEEN 20250101 AND 20250131\n AND mgame.sgamecode = 'initiatived'\n AND mgame.saccounttype = '-100'\n AND mgame.splattype IN ('-100', 'PC')\n AND mgame.splat = '-100'\n AND mgame.suseridtype IN ('qq', 'wxid');",
"result": [
{
"gplayerid": "app123"
}
]
},

BaseLine方案

官方给的baseline使用workflow实现，workflow看起来有点像绘制流程图并对模块进行编程。

本设计方案

基于LangGraph构建Multi-Agent,主体部分有SQL生成Agent、SQL执行与修正Agent、SQL选择Agent，将agent与工作流进行融合。

1. Agent 增强工作流
   在传统工作流中嵌入 Agent，可使某些步骤更具智能性。例如在文档审核工作流中，原本的人工审核步骤可引入智能审核 Agent。该 Agent 能够利用自然语言处理技术自动分析文档内容，检查语法错误、逻辑一致性等问题，并给出初步审核建议，大幅提高审核效率与准确性。
2. 工作流协调多个 Agent
   多个 Agent 可以被组织成一个工作流，各自承担不同的任务，通过工作流来协调它们之间的协作关系。比如在复杂的项目管理场景中，有负责需求分析的 Agent、负责资源分配的 Agent、负责进度跟踪的 Agent 等。通过工作流将这些 Agent 的工作有序编排，明确它们之间的输入输出关系和执行顺序，从而实现整个项目的高效管理。
3. 优势互补提升效率
   工作流提供了清晰的流程框架和稳定性，确保任务按照一定规则和顺序执行；而 Agent 则为系统带来了智能性和灵活性，能够处理复杂决策和环境变化。两者结合，可使系统既具有可预测性和可控性，又具备应对复杂情况的能力，进而提升整体工作效率和质量。例如在电商订单处理系统中，工作流负责管理从订单生成到发货的整体流程，而在其中的客服环节，可引入智能客服 Agent 来处理用户的咨询和投诉，实现了流程管理与智能服务的有机结合。

from langgraph.graph import StateGraph, START, END
from nodes import *
from state import GraphState
import sys
import os

# 获取项目根目录的绝对路径
current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.abspath(os.path.join(current_dir, '../..'))

# 添加项目根目录到Python路径
if project_root not in sys.path:
    sys.path.insert(0, project_root)

# 现在可以导入
# from ssy.langgraph_agent_V3_1.nodes import business_understanding_node


def create_multiagents_workflow():
    """创建多智能体工作流"""
    
    workflow = StateGraph(GraphState)

    # 添加所有节点
    workflow.add_node("business_understanding", business_understanding_node)
    workflow.add_node("sql_generation", sql_generation_node)
    workflow.add_node("sql_execution_node", sql_execution_node)
    workflow.add_node("sql_selection", sql_selection_node)

    # 定义工作流
    workflow.set_entry_point("business_understanding")

    # 单线连接
    workflow.add_edge(START, "business_understanding")
    workflow.add_edge("business_understanding", "sql_generation")
    workflow.add_edge("sql_generation", "sql_execution_node")
    workflow.add_edge("sql_execution_node", "sql_selection")

    def should_continue_iteration(state: GraphState) -> str:
        """判断是否继续迭代"""
        iteration_count = state["episode"]
        max_iterations = state["max_episode"]
        
        # 检查是否达到最大迭代次数
        if iteration_count >= max_iterations:
            print(f"🛑 达到最大迭代次数({max_iterations})，停止迭代")
            return "end"
        
        print(f"🔄 第{iteration_count + 1}轮迭代继续...")
        return "continue"
    # 条件边
    workflow.add_conditional_edges(
        "sql_selection",
        should_continue_iteration,  # 判断函数
        {
            "continue": "sql_generation",  # 继续生成
            "end": END         # 结束迭代
        }
    )

    return workflow.compile()


# 创建并导出工作流
workflow = create_multiagents_workflow()

核心模块解析

SQL 生成节点 (sql_generation_node)

• 异步并发生成：利用 asyncio.gather 并发调用多个模型（如 DeepSeek-Reasoner, Doubao-Thinking 等）以及不同温度参数（temperatures: [0, 0.188, 0.588]），一次性生成多个风格和思路的候选 SQL。
• 上下文注入：输入不仅包含表结构（DDL 转 Markdown），还包含了业务知识（knowledge）、通用规则（common_knowledge）、少量优质样本（golden_sql）以及用于辅助理解的插入数据（insert_data_info）。
• 迭代优化：如果这不是第一轮（episode > 0），会将上一轮裁判模型的修改建议（impromvements）注入 Prompt。

SQL 执行与自修正节点 (sql_execution_node)

• 真实执行引擎：调用 sql_exe.py 中的 execute_sql_with_pymysql 连接真实的 StarRocks/MySQL 数据库执行生成的 SQL。
• 内部修正循环 (Self-Correction)：
  • 系统检查每个候选 SQL 的执行状态。
  • 对于报错的 SQL（execution_success == False），系统会将错误信息和原 SQL 发送给 LLM 进行修正（SQL_CORRECTION_USER_PROMPT）。
  • 修正循环最多执行 MAX_FIX_NUMB = 5 次，极大提高了最终可用候选 SQL 的数量。

SQL 裁判选择节点 (sql_selection_node)

• 智能体评委：使用能力极强的模型作为裁判。
• 多维评估标准：
  • 执行成功性：屏蔽语法错误的 SQL。
  • Schema 依从性：严格检查字段名和表名。
  • 输出列匹配：要求 SELECT 的别名必须和问题要求的输出列完全一致。
  • 插入数据校验 (亮点)：通过比对真实插入测试数据（Insert Data）的查询结果，验证业务逻辑的合理性（避免“语法正确但逻辑荒谬”的 SQL）。
• 记录原因：选出最佳 SQL 并输出详细原因（selection_reason），存入 State 供下一轮迭代参考。

整体逻辑

“生成-执行-修正-评估”闭环：不仅依赖 LLM 的生成能力，更引入了编译器（DB引擎）的确定性反馈。这极大地提升了 Text-to-SQL 的准确率。

测试数据驱动 (Test-Driven)：将业务里的 insert_data 作为 Prompt 的一部分喂给模型，不仅用于生成，还用于评估阶段验证结果集是否符合常理，模拟了真实数据分析师的开发过程。

混合模型策略：生成阶段使用开源模型或低成本模型的高并发试错，评估阶段（裁判）使用昂贵但逻辑能力更强的模型进行拍板，兼顾了成本与效果。

Prompt 与 Context 工程

代码在构建上下文和提示词时，极大地丰富了信息密度，采用了非常标准的高阶 RAG（检索增强生成）与 Prompt 技巧。

1. 极致的上下文注入 (Context Density) 为了防止大模型“幻觉”，代码向 Prompt 中注入了极其丰富的真实业务上下文：

• 结构上下文：不仅告诉大模型表名（table_list），还通过 ddl_to_markdown 将 DDL 语句转换为 Markdown 格式（table_descriptions），清晰展示字段、类型和中文 Comment 注释。
• 数据上下文（核心亮点）：通过 insert_data_info 将真实的数据库测试数据（Insert 语句） 喂给大模型。这不仅帮助生成，更在裁判评估阶段（SQL_SELECTION_SYSTEM_PROMPT）让大模型能够根据真实数据的值域来判断 SQL 逻辑是否荒谬。
• 业务上下文：区分了特定问题的业务知识（knowledge）和全局通用的业务规则（common_knowledge），强约束模型的业务逻辑理解。

2. 少样本提示工程 (Few-Shot Prompting)

• 在生成和评估的 Prompt 中，都注入了 golden_sql（标准答案/优质样本）。
• 这不是简单的 QA 对，而是明确告诉大模型“参考以下成功的 SQL 模式”（如 CTE 的使用、时间范围的精确处理、 datediff 计算留存等），引导大模型模仿优秀的编码风格。

3. 动态上下文构建 (Dynamic Context Update)

• 在 sql_generation_node 中，如果进入了第二轮或之后的迭代（state["episode"] > 0），代码会动态提取上一轮裁判留下的历史修改建议： improvement_guidelines = "\n\n".join([f"{guideline_id+1}: \n{json.dumps(...)}"...])
• 这种将“历史失败教训”动态组合成新 Context 的做法，是强化大模型迭代学习能力的关键工程。

4. 强约束的输出格式 (Output Formatting Constraints)

• 在所有 Prompt 中，都通过 Markdown 格式和 JSON 示例严格定义了输出结构（例如 {"candidate": "SQL"} 或 {"selected_index": 0, "reasoning": {...}}）。
• 为了防止大模型在输出 JSON 时带有 Markdown 代码块包裹（json ... ），作者在 utils.py 中专门写了 extract_json 和 extract_answer（处理 <think> 标签）进行后处理工程，保证程序的鲁棒性。