当业务需要把自然语言问题直接转化为数据库查询结果时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/activate2.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( modelgpt-4o-mini, temperature0, ) 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 到 AgentLangChain 如何编排能力LangChain 早期版本的核心抽象是 Chain链把“提示词 → 模型 → 输出解析”串联起来。但随着应用复杂度提升单纯线性链无法应对“模型需要自己决定下一步调用什么工具”的场景于是 Agent 成了新的核心范式。Agent 的执行循环可以概括为接收用户输入。将输入和已有上下文交给大模型。模型决定直接回答或者调用某个工具。如果调用工具LangChain 会执行对应函数把结果作为“观察”返回给模型。重复步骤 2 到 4直到模型给出最终答案或达到最大迭代次数。这个循环就是 Harness 工程中最核心的部分。4. Harness 工程Agent 应用落地的工程化方法4.1 什么是 Harness 工程Harness 工程的核心目标是让不可控的大模型在可控的框架里工作。你可以把 Agent 理解为“决策大脑”而 Harness 就是承载大脑运行的“身体和护栏”。具体来说Harness 层需要提供以下能力能力说明工具注册与生命周期管理统一管理 Agent 可使用的工具包括工具描述、参数 schema、执行函数执行循环控制控制“模型→工具→模型”的循环设置最大迭代次数和终止条件超时与重试模型调用或工具执行超时时能优雅降级或重试权限校验在执行敏感操作前进行权限检查比如限制只读 SQL可观测性记录每次工具调用、模型输入输出的关键日志便于排查问题错误处理工具执行失败时将错误信息反馈给模型让它自行修正4.2 Harness 与 Agent 的区别不少开发者容易混淆 Harness 和 Agent。简单理解Agent 是业务逻辑层面的智能体它负责理解任务、选择工具、生成回复。Harness 是基础设施层面承载 Agent 的工程系统它不关心业务逻辑只负责让 Agent 稳定运行。用驾驶作比喻Agent 是司机Harness 是汽车。司机决定去哪里、怎么开汽车提供动力、刹车、仪表盘和碰撞保护。没有 HarnessAgent 就像没有安全带的司机能跑但风险很高。在 LangChain 生态中AgentExecutor就是一个基础 Harness而 LangGraph 则是更灵活的 Harness它允许你把 Agent 流程拆成节点Node和边Edge用图的方式来控制状态流转适合复杂的多智能体协作场景。4.3 工具调用循环的设计工具调用循环是 Harness 工程的核心机制。在 LangChain 中AgentExecutor内部自动处理了这个循环我们可以通过参数控制循环行为executor AgentExecutor( agentagent, toolstools, verboseTrue, max_iterations5, handle_parsing_errorsTrue, )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 typetable) tables [row[0] for row in cursor.fetchall()] schema_parts [] for table in tables: cursor.execute(fPRAGMA 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_asciiFalse) except Exception as e: return fSQL 执行出错{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( modelos.getenv(LLM_MODEL, gpt-4o-mini), temperature0, base_urlos.getenv(LLM_BASE_URL), api_keyos.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( agentagent, toolstools, verboseTrue, max_iterations5, handle_parsing_errorsTrue, ) return executor这里需要解释几个关键参数create_tool_calling_agent是 LangChain 中用于创建“工具调用型 Agent”的函数它要求模型支持 Function Calling / Tool Calling 能力。agent_scratchpad是 LangChain 内部机制用来保存中间的工具调用记录和观察结果。max_iterations5是为了避免模型陷入“生成 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_MODELgpt-4o-mini LLM_API_KEYyour_api_key # 如果你使用的是 OpenAI 兼容的网关或本地模型服务可以填写 base_url LLM_BASE_URL6.6 运行与验证在项目根目录执行python main.py输入“每个分类的平均价格是多少”后你会在终端看到 Agent 的工具调用过程模型调用get_db_schema获取表结构。模型生成 SQL。模型调用execute_sql执行查询。模型根据结果生成最终中文回答。预期输出类似 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.00verboseTrue会打印 Agent 内部的思考与工具调用过程开发阶段非常适合用来排查问题。7. 常见问题与排查思路7.1 工具调用超时the agent execution provider did not respond in time这是一个比较常见的报错核心意思是 Agent 在等待某个执行环节时超时了。出现这种报错的常见原因包括模型服务响应过慢尤其是高峰期。工具函数内部出现长时间阻塞比如 SQL 执行了超大表的全表扫描。网络波动导致 API 连接超时。排查步骤建议按顺序来先单独测试模型接口确认模型服务本身是否稳定。再单独调用每个工具函数确认工具执行耗时是否在可接受范围内。检查是否设置了合理的模型请求超时时间可以在ChatOpenAI中通过timeout参数控制。检查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 的verboseTrue适合开发阶段生产环境建议接入 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 安全边界”这套方法论可以复用到文档问答、数据看板、告警处理、流程审批等更多业务中。把工程化思维刻进骨子里才是大模型应用开发真正的分水岭。
当业务需要把自然语言问题直接转化为数据库查询结果时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/activate2.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( modelgpt-4o-mini, temperature0, ) 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 到 AgentLangChain 如何编排能力LangChain 早期版本的核心抽象是 Chain链把“提示词 → 模型 → 输出解析”串联起来。但随着应用复杂度提升单纯线性链无法应对“模型需要自己决定下一步调用什么工具”的场景于是 Agent 成了新的核心范式。Agent 的执行循环可以概括为接收用户输入。将输入和已有上下文交给大模型。模型决定直接回答或者调用某个工具。如果调用工具LangChain 会执行对应函数把结果作为“观察”返回给模型。重复步骤 2 到 4直到模型给出最终答案或达到最大迭代次数。这个循环就是 Harness 工程中最核心的部分。4. Harness 工程Agent 应用落地的工程化方法4.1 什么是 Harness 工程Harness 工程的核心目标是让不可控的大模型在可控的框架里工作。你可以把 Agent 理解为“决策大脑”而 Harness 就是承载大脑运行的“身体和护栏”。具体来说Harness 层需要提供以下能力能力说明工具注册与生命周期管理统一管理 Agent 可使用的工具包括工具描述、参数 schema、执行函数执行循环控制控制“模型→工具→模型”的循环设置最大迭代次数和终止条件超时与重试模型调用或工具执行超时时能优雅降级或重试权限校验在执行敏感操作前进行权限检查比如限制只读 SQL可观测性记录每次工具调用、模型输入输出的关键日志便于排查问题错误处理工具执行失败时将错误信息反馈给模型让它自行修正4.2 Harness 与 Agent 的区别不少开发者容易混淆 Harness 和 Agent。简单理解Agent 是业务逻辑层面的智能体它负责理解任务、选择工具、生成回复。Harness 是基础设施层面承载 Agent 的工程系统它不关心业务逻辑只负责让 Agent 稳定运行。用驾驶作比喻Agent 是司机Harness 是汽车。司机决定去哪里、怎么开汽车提供动力、刹车、仪表盘和碰撞保护。没有 HarnessAgent 就像没有安全带的司机能跑但风险很高。在 LangChain 生态中AgentExecutor就是一个基础 Harness而 LangGraph 则是更灵活的 Harness它允许你把 Agent 流程拆成节点Node和边Edge用图的方式来控制状态流转适合复杂的多智能体协作场景。4.3 工具调用循环的设计工具调用循环是 Harness 工程的核心机制。在 LangChain 中AgentExecutor内部自动处理了这个循环我们可以通过参数控制循环行为executor AgentExecutor( agentagent, toolstools, verboseTrue, max_iterations5, handle_parsing_errorsTrue, )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 typetable) tables [row[0] for row in cursor.fetchall()] schema_parts [] for table in tables: cursor.execute(fPRAGMA 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_asciiFalse) except Exception as e: return fSQL 执行出错{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( modelos.getenv(LLM_MODEL, gpt-4o-mini), temperature0, base_urlos.getenv(LLM_BASE_URL), api_keyos.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( agentagent, toolstools, verboseTrue, max_iterations5, handle_parsing_errorsTrue, ) return executor这里需要解释几个关键参数create_tool_calling_agent是 LangChain 中用于创建“工具调用型 Agent”的函数它要求模型支持 Function Calling / Tool Calling 能力。agent_scratchpad是 LangChain 内部机制用来保存中间的工具调用记录和观察结果。max_iterations5是为了避免模型陷入“生成 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_MODELgpt-4o-mini LLM_API_KEYyour_api_key # 如果你使用的是 OpenAI 兼容的网关或本地模型服务可以填写 base_url LLM_BASE_URL6.6 运行与验证在项目根目录执行python main.py输入“每个分类的平均价格是多少”后你会在终端看到 Agent 的工具调用过程模型调用get_db_schema获取表结构。模型生成 SQL。模型调用execute_sql执行查询。模型根据结果生成最终中文回答。预期输出类似 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.00verboseTrue会打印 Agent 内部的思考与工具调用过程开发阶段非常适合用来排查问题。7. 常见问题与排查思路7.1 工具调用超时the agent execution provider did not respond in time这是一个比较常见的报错核心意思是 Agent 在等待某个执行环节时超时了。出现这种报错的常见原因包括模型服务响应过慢尤其是高峰期。工具函数内部出现长时间阻塞比如 SQL 执行了超大表的全表扫描。网络波动导致 API 连接超时。排查步骤建议按顺序来先单独测试模型接口确认模型服务本身是否稳定。再单独调用每个工具函数确认工具执行耗时是否在可接受范围内。检查是否设置了合理的模型请求超时时间可以在ChatOpenAI中通过timeout参数控制。检查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 的verboseTrue适合开发阶段生产环境建议接入 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 安全边界”这套方法论可以复用到文档问答、数据看板、告警处理、流程审批等更多业务中。把工程化思维刻进骨子里才是大模型应用开发真正的分水岭。
