我让三个Agent替我写SQL、跑Python、出报告，结果差点把数据库删了

单Agent的幻觉让我多喝了五杯咖啡——复杂任务拆分是必然

事情要从帮一个朋友搞教育SaaS平台说起。他们日活大概5万用户，后台PostgreSQL里躺着学生答题记录、课程购买、教师评价，每天运营总监都会在群里问：“上个月完课率最高的10门课是哪些？”“找出最近一周付费转化突然下降的城市，可能是什么原因？”这类问题。最初我给他们搭了一个基于LangChain的单Agent系统，用的是ReAct模式加SQL工具，心想GPT-4直接查库、分析、写结论一条龙，多省事。

结果第一天就翻车了。一个简单的问题“上个月销量前三的课程，给出他们的平均评分”，Agent愣是跑了37步还没出结果。我看LangSmith的trace：它先查了销量表，得到了课程ID，然后又去查课程详情，接着又回去再查一遍销量——像是得了健忘症。更恶心的是，它在某一步生成了不带WHERE条件的聚合查询，把两千万行记录全扫了一遍，PG直接锁了30秒，前端超时报错。token消耗单次就到了18k，成本大约$0.2（那时候我还用的text-davinci-003，贵得要死）。

我折腾了一下午，先是加了详细的提示词约束：“每次查询必须带LIMIT 100”“不要重复执行已经完成的操作”“一次性获取所有必要字段避免回表”。情况好了一丁点儿，但面对多步骤推理任务（比如“分析过去30天完课率的变化趋势，并和上个月同期对比”），Agent还是会陷入循环或跳过关键步骤。后来我翻了不少论文和开源项目，意识到单个Agent的推理深度是有限的——它一边要规划步骤，一边要理解工具返回的数据，一边还要组织语言，上下文一长就丢失目标。这跟人一样，你不可能一边写SQL、一边做数据透视、一边写报告的同事还负责统筹全局。

所以我决定把系统拆成多个Agent，各自负责一个狭窄领域。这不是拍脑袋的决定，是我在跟单Agent搏斗了一周、喝了无数杯咖啡之后痛下的决心。下面这张表是我在决定拆分前记录的一些数据，足以说明单个巨型Agent的极限。

38%的成功率根本没法上线。就算勉强出来了，答案也可能是错的。而且单Agent的提示词越写越长，最后光system prompt就快1200 token，维护起来像在修改宪法。

多Agent架构设计：一个Coordinator和三个干活的

拆分成多个Agent的核心逻辑很简单：把“做计划”和“执行计划”分离。我设计了一个Coordinator（协调者）负责理解用户意图、制定任务步骤、调用子Agent，并在最后整合结果。三个子Agent分别是：SQL Agent（只会写SQL并解释结果）、Python Agent（执行数据分析代码，比如pandas操作、画图）以及Report Agent（把数据和结论组织成自然语言报告）。每个子Agent都是独立的LangChain Agent，本身也是一个小型ReAct loop，但他们的工具非常狭窄，基本不会出圈。

这种架构的好处是每个Agent的上下文非常干净，不会互相污染。Coordinator只看到“我需要从数据库里拿某张表的最近30天数据”这样一个高层次的任务，而不需要知道具体表结构；SQL Agent则看到具体的表schema，负责生成安全高效的查询。下面是一个简化版的架构图（我画在博客里了），整个系统通过自定义的LangChain Tool把子Agent包装成工具，Coordinator直接调用。

技术选型上我用了LangChain 0.1.11，搭配ChatOpenAI（gpt-4-0125-preview作为Coordinator的模型），子Agent用gpt-3.5-turbo-0125来节省成本。当然你可以全用GPT-4，但贵。我把子Agent封装成BaseTool，每个工具的描述极为关键，因为它决定了Coordinator会不会正确调用。

from langchain.tools import BaseTool
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

# 假设已经为每个子Agent定义了创建函数返回AgentExecutor
# 例如 create_sql_agent() 返回一个SQL AgentExecutor
sql_agent_executor = create_sql_agent()
python_agent_executor = create_python_agent()
report_agent_executor = create_report_agent()

class SQLAgentTool(BaseTool):
    name = "sql_agent"
    description = "执行数据库查询。输入应为自然语言的问题，该Agent会将其转换为SQL并执行。仅限只读查询。返回查询结果摘要。"
    
    def _run(self, query: str) -> str:
        return sql_agent_executor.invoke({"input": query})["output"]
    
    async def _arun(self, query: str) -> str:
        return await sql_agent_executor.ainvoke({"input": query})["output"]

# 类似定义 PythonAgentTool 和 ReportAgentTool
# ...

tools = [SQLAgentTool(), PythonAgentTool(), ReportAgentTool()]

# Coordinator的LLM
llm = ChatOpenAI(model="gpt-4-0125-preview", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个数据团队的协调者。你的职责是：
- 理解用户的数据分析需求
- 将复杂任务分解成子任务
- 使用可用的工具（sql_agent, python_agent, report_agent）完成每个子任务
- 在调用工具前，清晰地向工具描述需要做什么
- 整合所有工具的结果，给出最终答案
注意：
- 每个工具调用仅限一个明确的子任务
- 如果某工具返回错误，尝试一次其他路径或报告问题
- 不要猜测数据，所有数值必须来自工具
- 最终回答里要引用工具提供的具体数据
"""),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

coordinator_agent = create_openai_tools_agent(llm, tools, prompt)
coordinator_executor = AgentExecutor(agent=coordinator_agent, tools=tools, verbose=True)

关键点在于SQLAgentTool的description字段。如果描述写得太宽泛比如“处理数据”，Coordinator很容易把Python分析的活儿也扔给SQL Agent。我后来把描述精简成“仅限于只读查询”，并在system prompt里强调“SQL Agent只能查询数据库，不能做计算或复杂分析”，情况才好转。这个协调逻辑其实是一种“宏观ReAct”，Coordinator在循环中思考—>选择工具—>观察结果—>再思考，步数通常控制在3到6步之内，远少于单Agent的十几步。

整个系统的调用入口就是向coordinator_executor.invoke丢进去用户问题，它会自动编排三个Agent干活。下面是一次真实trace：用户问“过去7天新注册用户的完课率，按城市对比，并生成简要报告”。Coordinator第一步调用了SQL Agent去查新用户完课率按城市；第二步发现数据不够需要对比总完课率，又调了一次SQL；第三步让Python Agent计算差异并排序；第四步把两个Agent的结果扔给Report Agent生成文本。总共4步，耗时32秒，token大约7.2k。对比单Agent直接冲上去乱查，不仅快了一半，而且答案可信。

子Agent的定义和工具集成：我差点让SQL Agent把整个数据库删了

构建子Agent的过程不比Coordinator简单，因为每个子Agent内部也在用ReAct调用自己的工具。SQL Agent我用的是LangChain自带的create_sql_agent和SQLDatabaseToolkit，这玩意儿很方便，但是它默认可以执行任何SQL，包括DROP TABLE。我一开始偷懒没给数据库用户设置只读权限，心想反正测试库随便搞。结果有一次Agent生成的SQL里多了一个分号，后面跟了句DROP TABLE——PostgreSQL里多语句是被允许的（如果你用psycopg2），它真给我删了一张表。还好是测试环境，但吓得我马上给数据库用户加了只读权限，并且在SQL Agent的system prompt里明确写了“你只能执行SELECT语句，任何尝试修改数据库的操作都必须拒绝”。

from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import create_sql_agent
from langchain_openai import ChatOpenAI

db = SQLDatabase.from_uri("postgresql://readonly_user:pass@host/db")  # 只读用户
llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)

# 自定义前缀，严格限制
prefix = """
你是一个PostgreSQL专家。给定一个数据库，你可以创建只读SQL查询来回答用户问题。
数据库包含以下表：
{table_info}
重要规则：
1. 你只能执行SELECT查询。
2. 查询必须包含LIMIT 1000，除非用户明确要求所有数据。
3. 如果查询执行出错，分析错误并重写查询，最多重试2次。
4. 用中文解释查询结果。
"""

sql_agent = create_sql_agent(
    llm=llm,
    db=db,
    agent_type="openai-tools",
    prefix=prefix,
    verbose=True,
    max_iterations=5,
    early_stopping_method="force"
)

这个prefix是经过反复试错才定下来的。起初我没有限制LIMIT，结果一个查询返回了12万行，Agent上下文被塞满，直接报token超限。我把max token从4096调高到8k，但治标不治本。所以强制LIMIT是必须的，因为我们不需要把原始数据全部交给Agent，它只是需要概括数据。早期我还踩过一个坑：SQL Agent生成的查询里的列名偶尔会拼错，比如把user_id写成userId。因为我用的是PostgreSQL，列名小写加下划线，Agent有时候受自然语言污染。所以我后来又加了一条规则：“所有列名按数据库schema中的精确名称编写，严格遵循小写和下划线命名”。这样一来出错率下降了不少。

Python Agent相对更危险，因为它可以执行任意代码。我用了一个受限制的Python REPL工具，在子进程里沙箱化运行，限制最大执行时间10秒，禁止文件系统和网络访问。我参考了LangChain的PythonREPLTool但重写了执行部分，用subprocess.run并以timeout参数限制。

import subprocess
import tempfile

def execute_python_code(code: str) -> str:
    # 在临时文件中写入代码，然后用subprocess执行
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
        f.write(code)
        script_path = f.name
    try:
        result = subprocess.run(
            ['python3', script_path],
            capture_output=True, text=True, timeout=10, 
            env={"HOME": "/tmp"}  # 限制环境
        )
        if result.returncode != 0:
            return f"执行错误: {result.stderr}"
        # 只返回前2000字符，防止大输出撑爆上下文
        return result.stdout[:2000]
    except subprocess.TimeoutExpired:
        return "执行超时（超过10秒）"
    finally:
        # 清理临时文件
        import os
        os.unlink(script_path)

# 然后将这个函数包装成Tool就行

这种限制牺牲了灵活性，比如没法让Agent保存图片到磁盘，但我们的场景里Report Agent只需要文本描述，图片不是必需品。如果真需要画图，我会让Python Agent输出base64编码的图片，然后由应用层呈现，不经过LLM。

Report Agent更简单，它就是一个只有系统提示的大模型，输入是结构化数据和指令，输出是报告文本。我给它写了一个很长的模版，其中包含“你必须使用以下Markdown格式：标题、小标题、列表……”。为了让它不胡编数据，我把前面Agent返回的数据直接插在prompt里，并要求“只基于提供的数据陈述，不要添加未提及的信息”。这点效果还不错。

工具集成的核心教训是：工具描述决定了整个链条的可靠性。Coordinator依赖description来选择工具，一旦description不准确或模糊，就会错误调用。而且子Agent内部的工具如果功能重叠，也可能导致内循环。所以我在设计时尽量保持每个Agent的工具集最小化且正交。

协调器Prompt的迭代史：从三句话到一页纸，准确率从50%到95%

Coordinator的提示词我前后改了不下20版。最开始我用的就是上面代码里的那几行，觉得“计划-执行-整合”的逻辑很清晰。但测试时发现，Coordinator有时候会跳过Python Agent，直接用SQL Agent算聚合然后自己口算，结果算错；有时候会连续调用两次SQL Agent而不等待中间结果；还有时候在得到最终答案后，仍然坚持再调用一次工具，形成无意义的空转。

这些问题根源都在于prompt的约束不够。我加了几条强制规则，比如：“在获取所有必要数据之前，不得开始撰写最终答案”“如果你觉得自己已经可以回答，请直接输出最终答案，不要额外调用工具”“如果连续两次工具调用没有得到新信息，停止并报告”。但是Agent不一定遵守，尤其当它很“固执”的时候。

后来我从AutoGPT的prompt里借鉴了“任务分解清单”的思路，让Coordinator在第一步就输出一个明确的任务计划，然后后续步骤必须对照计划执行。这个思路用了一次few-shot示例，并且强制它用特定格式输出计划。修改后的prompt片段如下：

你是一个任务协调者。收到用户问题后，你必须遵循以下格式：

第一步：计划
列出解决该问题需要的步骤序号，每个步骤指明使用的工具（sql_agent/python_agent/report_agent）和简要目的。格式：
计划：
1. [sql_agent] 查询xxx
2. [python_agent] 分析xxx
...

第二步：逐步执行
使用工具执行计划中的每一步。每执行完一步，检查输出，必要时调整后续计划。

第三步：整合与回答
所有步骤完成后，调用report_agent生成最终报告，或者直接输出最终回答。

必须遵守：
- 在计划中，如果某个步骤不需要，就跳过对应的工具。
- 严格按照计划顺序执行，除非中间结果要求改变方案，那么修改计划并继续。
- 最终回答中必须包含具体数值和来源。

配上这个要求，Coordinator的行为变得可控多了。我会在第一次调用时就得到一份计划，如果计划不合理，我甚至可以在用户侧中断并修正。配合LangSmith我可以清楚看到计划的执行情况。准确率从早期的不到50%提升到了大概95%（这里的准确率是指最终答案逻辑正确、数据无误，我手动抽样了200个问题）。

有一个细节：我禁用了Coordinator的handle_parsing_errors，并且要求如果JSON解析错误（因为工具调用是function calling），就重试一次。否则偶尔会由于模型输出格式错误直接崩溃，这在生产环境是不可接受的。此外，我还加入了“如果工具返回了错误，Coordinator应该把错误信息包含在最终回答里，而不是假装一切正常”，这避免了幻觉。

Prompt的迭代没有银弹，就是在一次次观察到bad case后加补丁。最终版本prompt超过了800 token，虽然长，但逻辑严密。我原本担心这么长的system prompt会让模型分心，但实测gpt-4处理长提示的能力很强，并没有副作用。

性能优化：Token消耗砍了60%，响应时间从45秒到18秒

多Agent系统部署到测试环境后，用户体验很糟糕。一个中等复杂度的查询需要45秒以上，而且每次查询成本在$0.3到$0.5之间。运营总监抱怨“比我自己跑SQL还慢”。我不得不在保证结果质量的前提下做性能优化。优化前后的对比：

怎么做到的？核心是三点：减少重复调用、利用缓存、并行化。

我发现Coordinator经常重复问SQL Agent类似的问题，比如先查“过去7天的完课率”，然后觉得不够又查“过去7天的总学生数”，其实这两者可以合并在一个SQL里用CTE一次完成。我给SQL Agent加了一条建议：“如果Coordinator的问题可以合并成一个SQL，请使用子查询或CTE减少交互次数”。同时在Coordinator的prompt里鼓励它一次性把需要的所有数据描述给SQL Agent。这样减少了一次额外调用。

其次，我引入了语义缓存。很多分析问题是周期性的，比如“上周的完课率”几乎每天有人问，只是日期变了。我用Redis和文本嵌入（text-embedding-3-small）缓存了相似问题的最终答案。当新问题与缓存问题的余弦相似度>0.95时，直接返回缓存结果而不触发整个Agent链。这对于重复性高的场景效果拔群，命中率达到30%左右。

另外，Coordinator调用Python Agent和SQL Agent有时是独立的，比如需要查询数据和进行复杂的统计分析，它们之间没有依赖，完全可以并行。但是LangChain的AgentExecutor默认是串行执行工具调用的，因为它是在单个Thought之后选择一个工具。于是我在Coordinator内部实现了简单的并行：修改了plan_and_execute的思路，如果计划中有两个独立步骤，我让Coordinator一次性生成两个工具调用（function calling支持多个function_call），然后并行执行并反馈结果。这需要自定义Coordinator的逻辑，稍微绕了点，但效果明显。并行后，原本需要分别等待的两个步骤可以重叠，节省了网络往返时间。当然，并行执行受限于API并发限制，我申请了更高的rate limit，并且在客户端用信号量控制并发数。

我还把子Agent的模型从gpt-4换成了gpt-3.5-turbo，这大大降低了成本和延迟。SQL Agent用3.5-turbo生成SQL已经足够，因为数据库schema明确，逻辑固定。偶尔生成的SQL不够高效，但不会错。唯一的妥协是，遇到复杂分析时，SQL Agent可能会给出平庸的解释，但我们最后有Report Agent润色，所以用户几乎感觉不到。Python Agent也是一样，小模型对简单pandas操作没问题。这样总模型的吞吐量上去了，因为3.5-turbo比4-turbo快得多。

还有一个细节：流式输出。虽然整个多Agent链路的最终结果是完整的文本，但中间每个Agent的输出我可以流式展示给用户，让用户看到“正在查询数据库…”“正在分析数据…”的进度。我用了LangChain的StreamingStdOutCallbackHandler，并且配合WebSocket向前端推送中间日志。这样在18秒内用户一直有东西看，感知等待时间大大降低。

生产部署遇到的坑：并发、可观测性和Agent循环监控

开发完成只是第一步，部署上线那周我掉了不少头发。第一个问题是并发。我们用FastAPI提供了接口，多个运营同事同时提问。每个请求会创建一个AgentExecutor，它内部会多次调用OpenAI API。免费账号下gpt-4-turbo的RPM限制是500，但我们的并发峰值可能同时有15个请求，每个内部可能要调用4次，瞬间就超过限制了，导致429错误。我开始用简单的令牌桶算法限流，限制同时处理的请求数不超过5，其余排队。但这导致等待时间变长。最终我升级了OpenAI的usage tier，提高了RPM，并结合了在应用层的异步并发池。代码大概是这样的：

import asyncio
from asyncio import Semaphore

sem = Semaphore(5)  # 最多同时处理5个Agent任务

async def process_query(query: str):
    async with sem:
        result = await coordinator_executor.ainvoke({"input": query})
        return result["output"]

即使如此，偶尔还是会有429，所以我在ChatOpenAI的配置里加了重试机制，用tenacity库，最多重试3次，指数退避。这勉强稳住了。

第二个坑是Agent循环监控。Coordinator虽然设定最大步数，但还是可能在限制内做一些无用功。我在AgentExecutor的early_stopping_method上用了"generate"选项，当达到max_iterations时会强制用LLM生成最终答案。但这并不完美，有时生成的答案缺乏数据支撑。更致命的是，如果Agent在中间某一步产生了异常大的输出（比如SQL Agent返回了一个巨大的错误堆栈），就会让Coordinator的上下文过长，后续步数变得极慢。我写了一个中间件，在每次工具调用返回后检查输出长度，超过5000字符就截断并追加“[结果已截断]”。然后监控系统会报警，让我去排查是不是SQL错了。这个粗暴办法省了很多token和时间。

可观测性方面，我全面接入了LangSmith。每个请求作为一个trace，我可以看到所有嵌套的Agent调用、token消耗、延迟、工具输入输出。但是LangSmith的免费层限制trace数量，所以我只采样了20%的流量做详细trace。对于其余流量，我自定义了简单的日志记录关键指标到本地文件，再推送到Grafana。我强烈建议在生产环境搞一个dashboard，能实时看到平均步数、token使用趋势、错误率。有一次我发现错误率突然升高，检查后发现是数据库连接池被耗尽，SQL Agent拿到连接超时报错，但Coordinator没有正确处理这个错误，直接返回了“无法获取数据”。通过报警我们很快恢复了连接池。

还有一个哭笑不得的坑：Coordinator偶尔会自己发明工具调用参数。比如它想调用报告Agent，但参数里塞了一堆废话，导致子Agent返回无关内容。这是function calling的已知问题。我把子Agent的tool schema里的参数描述写得更苛刻，并且限制了参数长度。同时，在Coordinator的prompt中明确：“调用工具时，参数只写问题的简洁描述，不要包含多余的解释。”有所好转，但没法完全杜绝。

最终这个系统在生产环境跑了一个多月，处理了大概2300个查询。平均响应时间17.5秒，成功率（最终答案被用户点赞或默认为有用）大约87%。虽然没达到99%，但对于一个非关键线业务的数据助手来说已经够用。朋友的公司省了至少一个初级数据分析师的人力，每天自动生成晨报，管理层挺满意。

我还在持续迭代，比如尝试用LangGraph重构协调逻辑以获得更灵活的并行和条件路由，那是另一个故事了。如果你也在搞多Agent，我的核心建议就是：别想用一个巨人做所有事，拆分+清晰分工+监控，比什么都管用。