LangChain Agent实战：用Harness工程构建TextToSQL

当业务需要把自然语言问题直接转化为数据库查询结果时，TextToSQL 是性价比很高的一条技术路线。但真正落地时你会发现，单纯让大模型生成 SQL 远远不够：模型要理解表结构、要调用工具去执行查询、还要根据执行报错自我修正，整个过程需要一套稳定的工程框架来“托底”。这篇文章就从 LangChain 生态出发，围绕 Agent 开发、Harness 工程、智能体工具和 TextToSQL 项目完整展开，适合想入门大模型 Agent 开发、或者正在做数据问答类产品的开发者。

1. 背景：为什么 Agent 开发需要“工程化”

1.1 从大模型 API 到 Agent 应用

大模型提供了一个非常强大的能力：理解和生成自然语言。但只靠模型本身，我们很难让它完成“查询数据库并返回结果”这类真实业务操作。原因很简单：模型没有权限去连数据库、没有工具去执行 SQL、也不会在 SQL 报错后自动尝试修正。

于是 Agent 的概念出现了。Agent 可以理解为“有目标、能拆解任务、会调用工具、能根据反馈调整行动”的智能体。它不再只是“一问一答”的对话机器人，而是一个可以自主完成多步任务的工作单元。

但 Agent 也不是银弹。把 Agent 接入生产环境后，你会遇到一连串问题：模型调用工具超时怎么办？模型循环调用工具停不下来怎么办？工具权限怎么控制？调用日志怎么记录？这些问题如果只靠脚本堆砌，项目会越来越难维护。

1.2 LangChain 在 Agent 生态中的定位

LangChain 是目前大模型应用开发中最常用的编排框架之一。它提供了统一的模型调用接口、提示词管理、工具抽象、Agent 执行循环等能力。简单说，有了 LangChain，你不需要自己从零实现“调用模型 → 解析工具调用 → 执行工具 → 把结果返回给模型”这套循环逻辑。

很多资料中提到的 LangChain V1.0，并不是指某个突然发布的独立大版本号，而是指 LangChain 经历多年迭代后形成的一套完整工具链：以langchain-core作为核心抽象层，把模型、提示词、工具、记忆、Agent 执行器分层解耦，同时通过 LangGraph 这类库来处理更复杂的图状态编排。对于开发者来说，更重要的是理解这套架构思想，而不是纠结具体包版本。

1.3 Harness 工程要解决什么问题

“Harness” 在英文里有“安全带、缰绳、控制装置”的意思。在大模型 Agent 开发中，Harness 工程可以理解为“为了让 Agent 可靠运行而构建的整套工程基础设施”。

如果没有 Harness 层，Agent 代码会变成一团乱麻：

• 模型的返回结果不稳定，有时正常返回文本，有时输出 JSON，有时甚至胡言乱语。
• 工具调用没有超时控制，遇到模型服务响应慢，整个请求可能一直挂起。
• 工具权限没有边界，模型可能被诱导执行危险的数据库操作。
• 没有日志和链路追踪，出问题后根本不知道 Agent 内部发生了什么。

Harness 工程要解决的正是这些问题：为 Agent 提供稳定可控的运行环境、工具注册与权限管理、执行循环控制、超时重试机制、可观测性和审计能力。

2. 环境准备与版本说明

2.1 基础运行环境

本文实战项目以 Python 3.10 或更高版本为例。操作系统方面，Windows、macOS、Linux 都可以，重点是确保命令行工具可用。

建议你创建独立的虚拟环境，避免依赖冲突：

python -m venv .venv # Windows .venv\Scripts\activate # macOS / Linux source .venv/bin/activate

2.2 项目依赖

本文示例使用 SQLite 作为演示数据库，好处是零配置、单文件，适合学习阶段快速验证。大模型部分，示例代码采用 OpenAI 兼容接口进行调用，你可以根据自己实际可访问的模型服务调整base_url、api_key和model参数。

依赖清单如下：

pip install langchain langchain-openai langchain-core python-dotenv

版本方面，LangChain 的 API 更新比较频繁，本文示例以当前可稳定使用的 API 写法为准。如果你使用的是较新版本，个别类名或函数入参可能有调整，遇到报错时优先查看官方最新文档。

2.3 项目结构规划

TextToSQL 项目并不复杂，但为了体现工程化思想，我们按职责拆分为数据层、工具层、提示词层和 Agent 层：

text2sql_demo/ ├── data/ │ └── create_db.py ├── src/ │ ├── __init__.py │ ├── tools/ │ │ ├── __init__.py │ │ ├── schema_tool.py │ │ └── sql_executor.py │ ├── prompts/ │ │ ├── __init__.py │ │ └── sql_prompt.py │ └── agent/ │ ├── __init__.py │ └── text_to_sql_agent.py ├── main.py ├── requirements.txt └── .env.example

这样拆分的好处是：工具可以单独测试，提示词可以单独调整，Agent 编排逻辑保持精简。后面扩展新功能时，只需要新增一个工具文件即可。

3. LangChain 核心概念与基础用法

3.1 模型调用：LangChain 如何统一大模型接口

LangChain 将不同厂商的大模型统一封装成ChatModel接口。开发者只需要关注两个核心方法：invoke用于单次对话，stream用于流式输出。

下面是最简单的一个示例：

from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="gpt-4o-mini", temperature=0, ) response = llm.invoke("用一句话解释什么是 Agent") print(response.content)

这段代码的本质是：LangChain 帮你把消息转换成模型需要的格式，再返回统一的结构化结果。对于 TextToSQL 这类任务，temperature建议设为 0，因为 SQL 生成任务需要确定性，不希望模型“发挥创造力”。

3.2 PromptTemplate：提示词管理

在代码里直接拼接提示词会导致维护困难，LangChain 提供了ChatPromptTemplate来管理结构化提示词。

from langchain_core.prompts import ChatPromptTemplate prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个{role}，请根据用户问题完成SQL生成与执行。"), ("human", "{input}"), ]) filled = prompt.invoke({ "role": "数据分析助手", "input": "统计每个分类的平均价格", }) print(filled)

这里要注意，ChatPromptTemplate的变量用{变量名}占位，invoke时传入字典即可。在 Agent 开发中，提示词模板还需要额外的占位符来传递工具调用中间过程，后面会看到。

3.3 从 Chain 到 Agent：LangChain 如何编排能力

LangChain 早期版本的核心抽象是 Chain（链），把“提示词 → 模型 → 输出解析”串联起来。但随着应用复杂度提升，单纯线性链无法应对“模型需要自己决定下一步调用什么工具”的场景，于是 Agent 成了新的核心范式。

Agent 的执行循环可以概括为：

1. 接收用户输入。
2. 将输入和已有上下文交给大模型。
3. 模型决定直接回答，或者调用某个工具。
4. 如果调用工具，LangChain 会执行对应函数，把结果作为“观察”返回给模型。
5. 重复步骤 2 到 4，直到模型给出最终答案或达到最大迭代次数。

这个循环就是 Harness 工程中最核心的部分。

4. Harness 工程：Agent 应用落地的工程化方法

4.1 什么是 Harness 工程

Harness 工程的核心目标是：让不可控的大模型在可控的框架里工作。你可以把 Agent 理解为“决策大脑”，而 Harness 就是承载大脑运行的“身体和护栏”。

具体来说，Harness 层需要提供以下能力：

4.2 Harness 与 Agent 的区别

不少开发者容易混淆 Harness 和 Agent。简单理解：

• Agent 是业务逻辑层面的智能体，它负责理解任务、选择工具、生成回复。
• Harness 是基础设施层面承载 Agent 的工程系统，它不关心业务逻辑，只负责让 Agent 稳定运行。

用驾驶作比喻：Agent 是司机，Harness 是汽车。司机决定去哪里、怎么开；汽车提供动力、刹车、仪表盘和碰撞保护。没有 Harness，Agent 就像没有安全带的司机，能跑但风险很高。

在 LangChain 生态中，AgentExecutor就是一个基础 Harness，而 LangGraph 则是更灵活的 Harness：它允许你把 Agent 流程拆成节点（Node）和边（Edge），用图的方式来控制状态流转，适合复杂的多智能体协作场景。

4.3 工具调用循环的设计

工具调用循环是 Harness 工程的核心机制。在 LangChain 中，AgentExecutor内部自动处理了这个循环，我们可以通过参数控制循环行为：

executor = AgentExecutor( agent=agent, tools=tools, verbose=True, max_iterations=5, handle_parsing_errors=True, )

max_iterations控制模型最多调用几次工具，防止模型陷入死循环。handle_parsing_errors表示模型输出格式解析失败时，把错误重新交给模型处理。

这里特别提醒：在真实项目中，不要把所有工具都无脑暴露给 Agent。工具越多，模型选错工具的概率越大。Harness 层应该根据业务场景动态决定开放哪些工具。

5. 智能体工具：从 Function Calling 到自定义工具

5.1 工具（Tool）的本质

在 LangChain 中，工具是一个带有描述信息的可调用对象。工具描述非常重要，因为大模型正是根据“工具名称 + 工具描述”来决定何时调用、为何调用。

工具的本质结构包含三部分：

• 名称：在 Agent 系统中唯一，尽量清晰准确。
• 描述：说明工具能做什么、什么时候用。
• 参数：调用该工具需要传入的结构化参数。

5.2 使用@tool装饰器自定义工具

LangChain 提供了@tool装饰器，可以直接把函数变成工具。下面是一个获取数据库表结构的工具：

# 文件路径：src/tools/schema_tool.py import sqlite3 from langchain_core.tools import tool DB_PATH = "shop.db" @tool def get_db_schema() -> str: """获取 shop.db 数据库的全部表结构和字段类型，用于生成 SQL 语句。""" conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") tables = [row[0] for row in cursor.fetchall()] schema_parts = [] for table in tables: cursor.execute(f"PRAGMA table_info({table})") columns = cursor.fetchall() col_desc = ", ".join([f"{col[1]} {col[2]}" for col in columns]) schema_parts.append(f"表 {table} 字段: {col_desc}") conn.close() return "\n".join(schema_parts)

注意，函数的第一行注释就是工具描述。命令越清晰，模型越容易理解工具的用途。如果描述模糊，模型可能在一个问题上反复调用错误工具。

5.3 工具调用的错误处理与安全边界

工具执行失败是常态。例如 SQL 语法错误、数据库连接超时、字段不存在等。Harness 层要做的是：把错误信息返回给模型，让模型基于错误信息调整策略。

再来看执行 SQL 的工具：

# 文件路径：src/tools/sql_executor.py import json import sqlite3 from langchain_core.tools import tool DB_PATH = "shop.db" @tool def execute_sql(sql: str) -> str: """在 shop.db 上执行 SQL 只读查询（仅支持 SELECT），返回 JSON 格式的查询结果。""" sql_clean = sql.strip() if not sql_clean.upper().startswith("SELECT"): return "错误：当前工具仅支持 SELECT 查询，不允许执行写操作。" try: conn = sqlite3.connect(DB_PATH) conn.execute("PRAGMA query_only = ON") cursor = conn.cursor() cursor.execute(sql_clean) col_names = [desc[0] for desc in cursor.description] rows = cursor.fetchall() conn.close() # 只返回前 20 行，避免结果过大挤占模型上下文 limited = [list(map(str, row)) for row in rows[:20]] return json.dumps({"columns": col_names, "rows": limited}, ensure_ascii=False) except Exception as e: return f"SQL 执行出错：{e}"

上面有两个关键细节：

第一，通过PRAGMA query_only = ON强制数据库只读，即使模型生成了DELETE或UPDATE，数据库层面也会拒绝执行。这是 TextToSQL 项目最重要的安全护栏。

第二，查询结果只取前 20 行。如果不限制结果集大小，一旦表数据量很大，返回内容可能会超出模型的上下文窗口，导致后续调用失败。

6. TextToSQL 项目实战：完整落地流程

6.1 需求分析

我们要实现的能力是：用户输入一句中文问题，系统自动完成“理解问题 → 查看表结构 → 生成 SQL → 执行查询 → 返回中文答案”的完整流程。

需求拆解如下：

• 支持查询商品、订单等基础表。
• Agent 必须第一步先获取数据库 schema，才能生成有效 SQL。
• SQL 只能执行 SELECT，任何写操作都被拒绝。
• 最终答案要求用自然语言描述。

6.2 初始化 SQLite 数据库

先准备演示数据。创建一个create_db.py脚本，生成商品表和订单表：

# 文件路径：data/create_db.py import os import sqlite3 DB_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "shop.db") def init_database(): if os.path.exists(DB_PATH): os.remove(DB_PATH) conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute(""" CREATE TABLE products ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, category TEXT, price REAL NOT NULL, stock INTEGER DEFAULT 0 ) """) cursor.execute(""" CREATE TABLE orders ( id INTEGER PRIMARY KEY, product_id INTEGER NOT NULL, quantity INTEGER NOT NULL, order_date TEXT NOT NULL, customer_name TEXT, FOREIGN KEY (product_id) REFERENCES products(id) ) """) products = [ ("智能手环", "智能穿戴", 249.0, 120), ("蓝牙耳机Pro", "数码配件", 399.0, 80), ("机械键盘87键", "外设", 329.0, 60), ("USB-C扩展坞", "数码配件", 199.0, 150), ("智能台灯", "家居", 159.0, 90), ] cursor.executemany( "INSERT INTO products (name, category, price, stock) VALUES (?, ?, ?, ?)", products, ) orders = [ (1, 2, "2025-06-01", "张三"), (2, 1, "2025-06-01", "李四"), (3, 3, "2025-06-02", "王五"), (1, 1, "2025-06-03", "赵六"), (4, 2, "2025-06-04", "张三"), (2, 1, "2025-06-05", "钱七"), (5, 4, "2025-06-06", "孙八"), (3, 2, "2025-06-07", "李四"), ] cursor.executemany( "INSERT INTO orders (product_id, quantity, order_date, customer_name) VALUES (?, ?, ?, ?)", orders, ) conn.commit() conn.close() print(f"数据库初始化完成：{DB_PATH}") if __name__ == "__main__": init_database()

在项目根目录执行：

python data/create_db.py

执行后会在项目根目录生成shop.db文件。

6.3 编写系统提示词

提示词是 TextToSQL 项目效果的关键。我们需要在系统提示词中明确 Agent 的工作步骤：

# 文件路径：src/prompts/sql_prompt.py SQL_SYSTEM_PROMPT = """ 你是一个智能数据查询助手。请将用户的自然语言问题转换为 SQL 查询，并利用工具执行查询。 请严格按以下顺序工作： 1. 首先调用 get_db_schema 工具查看数据库的表结构，了解有哪些表和字段。 2. 根据表结构和用户问题，编写一条只读 SELECT 查询语句。 3. 调用 execute_sql 工具执行这条 SQL。 4. 根据 SQL 执行结果，用简洁准确的中文回答用户。 补充要求： - 只允许使用 SELECT 查询，禁止生成 INSERT、UPDATE、DELETE、DROP 等写操作。 - 如果 SQL 执行报错，根据错误信息修正后重新执行，最多重试 2 次。 - 查询结果超过 10 行时，进行聚合统计或直接取前 10 行，不要返回冗长明细。 - 金额统一保留两位小数。 - 回答时说明数据来源表，方便用户确认。 """

这段提示词有几个设计要点：

• “首先调用 get_db_schema” 是一个强约束，能显著降低模型凭空猜测表名和字段名的概率。
• 明确“只允许 SELECT”和“最多重试 2 次”，防止模型陷入无限修正循环。
• 要求回答时说明数据来源表，让结果具备可追溯性。

6.4 构建 Agent 执行流程

把工具、提示词和模型组合成 Agent：

# 文件路径：src/agent/text_to_sql_agent.py import os from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from src.prompts.sql_prompt import SQL_SYSTEM_PROMPT from src.tools.schema_tool import get_db_schema from src.tools.sql_executor import execute_sql def build_agent(): llm = ChatOpenAI( model=os.getenv("LLM_MODEL", "gpt-4o-mini"), temperature=0, base_url=os.getenv("LLM_BASE_URL"), api_key=os.getenv("LLM_API_KEY"), ) tools = [get_db_schema, execute_sql] prompt = ChatPromptTemplate.from_messages([ ("system", SQL_SYSTEM_PROMPT), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ]) agent = create_tool_calling_agent(llm, tools, prompt) executor = AgentExecutor( agent=agent, tools=tools, verbose=True, max_iterations=5, handle_parsing_errors=True, ) return executor

这里需要解释几个关键参数：

• create_tool_calling_agent是 LangChain 中用于创建“工具调用型 Agent”的函数，它要求模型支持 Function Calling / Tool Calling 能力。
• agent_scratchpad是 LangChain 内部机制，用来保存中间的工具调用记录和观察结果。
• max_iterations=5是为了避免模型陷入“生成 SQL → 出错 → 再生成”的无限循环。

6.5 编写入口程序

最后写一个交互式入口，让它支持命令行提问：

# 文件路径：main.py import os from dotenv import load_dotenv load_dotenv() from src.agent.text_to_sql_agent import build_agent def main(): agent = build_agent() print("TextToSQL Agent 已启动，输入 exit 退出") print("示例问题：") print(" 1. 每个分类的平均价格是多少？") print(" 2. 销量最高的商品是哪个？") print(" 3. 张三一共下了几单？") while True: question = input("\n请输入你的问题：") if question.strip().lower() in ("exit", "quit"): break result = agent.invoke({"input": question}) print("\n回答：", result["output"]) if __name__ == "__main__": main()

环境变量文件.env.example：

# 模型接口配置，按你的实际部署情况填写 LLM_MODEL=gpt-4o-mini LLM_API_KEY=your_api_key # 如果你使用的是 OpenAI 兼容的网关或本地模型服务，可以填写 base_url LLM_BASE_URL=

6.6 运行与验证

在项目根目录执行：

python main.py

输入“每个分类的平均价格是多少？”后，你会在终端看到 Agent 的工具调用过程：

1. 模型调用get_db_schema，获取表结构。
2. 模型生成 SQL。
3. 模型调用execute_sql执行查询。
4. 模型根据结果生成最终中文回答。

预期输出类似：

> Entering new AgentExecutor chain... Invoking: `get_db_schema` ... Invoking: `execute_sql` SELECT category, ROUND(AVG(price), 2) as avg_price FROM products GROUP BY category; 回答：各分类的平均价格如下： - 智能穿戴：249.00 - 数码配件：299.00 - 外设：329.00 - 家居：159.00

verbose=True会打印 Agent 内部的思考与工具调用过程，开发阶段非常适合用来排查问题。

7. 常见问题与排查思路

7.1 工具调用超时：the agent execution provider did not respond in time

这是一个比较常见的报错，核心意思是 Agent 在等待某个执行环节时超时了。出现这种报错的常见原因包括：

• 模型服务响应过慢，尤其是高峰期。
• 工具函数内部出现长时间阻塞，比如 SQL 执行了超大表的全表扫描。
• 网络波动导致 API 连接超时。

排查步骤建议按顺序来：

1. 先单独测试模型接口，确认模型服务本身是否稳定。
2. 再单独调用每个工具函数，确认工具执行耗时是否在可接受范围内。
3. 检查是否设置了合理的模型请求超时时间，可以在ChatOpenAI中通过timeout参数控制。
4. 检查max_iterations是否设置过小，导致 Agent 在多次工具调用后达到上限而中断。

如果生产环境经常出现该类超时，建议在 Harness 层增加异步任务队列，把同步超时改造为异步任务状态查询。

7.2 模型生成的 SQL 不符合预期

常见表现：模型没有先查看表结构就直接生成了 SQL，导致表名或字段名错误。

解决方案：

• 强化系统提示词，明确要求“第一步必须调用 get_db_schema”。
• 在执行 SQL 前，让 Agent 先把生成的 SQL 展示出来，人工确认后再执行。
• 针对高频问题准备 few-shot 示例，在提示词中补充“用户问题 + 标准 SQL”的参考对。

7.3 Schema 上下文过大导致 Token 超限

业务数据库可能有很多表，完整的PRAGMA table_info信息非常长，会占用大量上下文窗口。

解决方案：

• 在get_db_schema工具中增加“按关键字过滤”的功能，只返回与问题相关的表结构。
• 对 schema 信息做压缩，只保留表名、字段名和注释，去掉字段类型中的长度信息。
• 如果确实需要全量 schema，建议使用支持长上下文的模型，但成本会更高。

7.4 SQL 注入与大模型越权风险

虽然我们限制了SELECT，但大模型生成的 SQL 仍然可能读取超出权限范围的数据。建议接入数据库账号体系，创建只拥有必要表 SELECT 权限的专用账号，让所有 Agent 查询都走这个账号，从数据库层面做权限隔离。

8. 最佳实践与工程建议

8.1 提示词与上下文管理

TextToSQL 类任务对确定性要求极高，temperature必须设为 0。同时，不要把大段业务规则塞进系统提示词，而是把规则拆分到工具层。例如“不允许删除数据”这类约束，最好的实现方式不是在提示词里反复强调，而是在 SQL 执行工具里通过query_only强制兜底。

8.2 安全与权限设计

涉及数据库操作的项目，安全设计是第一位。建议按以下层次叠加防护：

• 数据库层：创建只读账号，仅授予需要的表的SELECT权限。
• 工具层：在执行前校验 SQL 类型，非SELECT直接拒绝。
• 应用层：记录完整的查询日志，包括用户问题、生成的 SQL、执行结果、耗时。
• 审计层：对敏感表的数据查询进行额外审批和留痕。

8.3 可观测性与日志

Agent 项目排障最大的难点是“黑盒”。模型内部在想什么、为什么调用某个工具、工具返回了什么，这些都需要记录下来。LangChain 的verbose=True适合开发阶段，生产环境建议接入 LangSmith 等观测平台，或者至少自己实现一个回调函数，把关键事件输出到日志系统。

一个简单的日志回调示例：

from langchain_core.callbacks import BaseCallbackHandler class AgentLogCallback(BaseCallbackHandler): def on_tool_start(self, serialized, input_str, **kwargs): print(f"调用工具: {serialized.get('name')}，输入: {input_str}") def on_tool_end(self, output, **kwargs): print(f"工具返回: {output[:200]}")

8.4 成本控制

Agent 应用的成本比普通 API 调用高得多，因为一次任务可能需要多次模型调用。控制成本的方法包括：

• 限制max_iterations，避免无效循环。
• 工具返回结果做截断，减少上下文长度。
• 优先使用小模型处理简单查询，只有复杂场景才升级到更大模型。
• 对相同或相似问题进行结果缓存。

8.5 从 Demo 到生产的差距

本文项目是一个可运行的 Demo，但距离生产还有一段路要走。生产环境至少还需要补充：

• 数据库连接池，而不是每次查询都新建连接。
• 对用户输入进行安全过滤，防止提示词注入。
• 将工具执行能力封装成独立的微服务，与模型编排层解耦。
• 建立回归测试集，用一组标准问题持续验证 Agent 输出质量。

9. 总结与学习路线

这篇文章从 LangChain 的核心概念出发，介绍了 Harness 工程在 Agent 落地中的作用，并用一个完整的 TextToSQL 项目串起了工具定义、提示词设计、Agent 执行循环和安全边界设计。建议你按下面的顺序继续深入学习：

第一，把本文的 TextToSQL 跑通，然后尝试增加新的表，体会模型在 schema 变化后的表现。第二，深入学习 LangGraph，理解它与 AgentExecutor 在编排能力上的差异，尝试用图的方式重写本次 Agent。第三，研究 Function Calling 的原理，理解模型如何从文本生成结构化的工具调用参数。第四，尝试接入 RAG，让 Agent 在回答前先检索相关知识，进一步提升复杂问题的处理能力。

TextToSQL 只是 Agent 落地的一个典型场景，它背后的“模型 + 工具 + Harness + 安全边界”这套方法论，可以复用到文档问答、数据看板、告警处理、流程审批等更多业务中。把工程化思维刻进骨子里，才是大模型应用开发真正的分水岭。