拓冰网站定制 
最近半年大模型 Agent 已经从一个偏研究向的概念迅速变成了简历上的关键词、面试必问题和企业立项书里的常客。但真正动手写过 Agent 的开发者大多会有同一个感受Demo 跑起来很快一上生产就各种崩。LangChain 的 API 变来变去、LangGraph 的状态图设计要重新学一遍、Tool 调用时模型经常“假装调用”但输出格式不对、好不容易跑通了结果 SQL 生成出来根本不能执行。这些坑几乎每个 Agent 开发人员都会遇到。而如果你关注行业热点还会看到一个新的高频词——Harness 工程。它到底是什么跟 Agent 开发有什么关系这篇文章不是一份简单的新特性播报。我会用一条完整的TextToSQL 项目落地路径把 LangChain V1.0 的编排思路、LangGraph 的状态流设计、Agent 工具链的写法、以及 Harness 工程对生产可用性的价值串起来讲清楚。读完你会明白为什么 Agent 项目不能只靠“堆 Prompt 调 API”以及一个能稳定运行的智能体系统工程结构到底长什么样。1. 这篇文章真正要解决的问题先给一个明确判断当前大模型 Agent 开发的瓶颈不再是模型能力而是工程能力。模型能理解多少、能生成多好已经高度收敛但“能不能稳定执行”“失败后能不能自愈”“工具调用错不乱来”这些工程问题才是拉开项目差距的地方。如果你正在做以下事情这篇文章就是写给你的你已经在调大模型 API想从“单轮问答”升级到“能调用工具的 Agent”你被 LangChain 老版本的一堆 chain API 弄晕了想搞清楚 V1.0 到底该学什么你想做 TextToSQL / NL2SQL 项目但发现模型生成的 SQL 经常语法错、执行错、甚至查出和问题无关的数据你听说了 Harness 工程但不知道它在大模型项目里到底落在哪一层你的 Agent 经常出现 “agent execution terminated due to error” 这类问题想知道从哪里排查。这篇文章会以 TextToSQL 为例横跨四个层级来讲模型层怎么选与调、编排层用什么框架、工具层怎么设计、工程层怎么保证稳定。最终代码会跑通一个完整的“自然语言查数据库”最小闭环。2. 核心概念LangChain、LangGraph、Agent 与 Harness 工程很多初学者会把 LangChain 和 LangGraph 混为一谈这是第一个要纠正的认知。LangChain 是一个大模型应用编排框架。它的核心价值是帮你统一管理大模型调用、Prompt 模板、输出解析、外部工具调用等一堆琐碎的事情。你可以在 LangChain 里不用自己写繁琐的 HTTP 封装和 JSON 解析直接定义模型、工具和流程。不过LangChain 早期的 Chain API 抽象层次偏重很多人学过之后真正写项目时反而觉得不如直接用原生代码舒服。LangGraph 是面向“有状态、多步骤、有条件判断”的 Agent 流程编排框架。你可以把它理解成一个“把 Agent 流程画成流程图”的开发库。Node 是流程中的每一个步骤Edge 是步骤之间的跳转条件State 是在整个流程中流动的数据对象。LangGraph 最大的价值在于Agent 的执行不再是“一次调用就结束”而是可以循环模型决定调用工具 → 工具返回结果 → 模型继续分析 → 再决定调用下一个工具。这种循环在原生代码里写起来很容易乱用 LangGraph 则清晰很多。Agent 是指能自主规划任务、调用外部工具、并根据工具结果继续行动的智能体程序。跟普通大模型应用的关键区别在于普通应用是“用户问一句模型答一句”Agent 则是一个循环系统它有感知读取工具输入、决策判断下一步做什么、行动调用工具、观察读取工具结果四个阶段。Harness 工程是大模型 Agent 开发中一个很新的工程化概念。英文 “test harness” 在软件工程里指测试夹具/测试框架而在大模型领域Harness 工程指的是“把模型能力包裹进一套可控制、可验证、可重试、可观测的运行框架中”。说得直白一点Harness 就是 Agent 的执行安全壳。它负责管理请求重试、失败降级、输出校验、权限控制、日志追踪等与模型能力无关、但决定系统能否上线的脏活累活。传统应用开发里你从来不会让用户直接执行一段不可控的 SQL同样在你写 Agent 时你也不应该让模型直接裸调数据库、裸发 HTTP 请求。Harness 工程要解决的就是这个“接口可信”的问题。下面用一个表格对比它们的定位概念定位类比关键问题LangChain基础编排框架工具装箱统一 API怎么管理 Model / Prompt / ToolLangGraph流程状态编排流程图 状态机Agent 多步执行时如何跳转Agent智能体应用范式会思考的行动体怎么决策、怎么调用工具Harness 工程生产级运行外壳测试框架 安全壳失败重试、输出校验、日志、权限注意这四个概念不是并列的技术而是不同层的产物。LangChain 和 LangGraph 是“底座”Agent 是“模式”Harness 工程是“生产化的封装方式”。这也解释了为什么 LangChain 官方在 V1.0 之后重点推 LangGraph——因为单纯的“链式串联”已经满足不了 Agent 的复杂流程控制需求。3. Agent 开发为什么难从 Demo 到可用的鸿沟很多人写 Agent 的起点是照着一个 LangChain 官方示例改出来的。示例里有一个 ReAct Agent模型自己能看到工具描述、自己决定是否调用跑一个简单计算题或搜索题看起来真的“智能”。于是你想那我把数据库查询、接口调用都封装成工具模型不就能帮我干活了吗真实情况是一旦把 Agent 接入真实业务问题立刻涌出来第一模型输出不稳定。模型“决定调用工具”的方式是输出一个结构化的工具调用对象。但真实项目中你可能会遇到返回了空字符串、返回了拼写错误的方法名、返回了根本不存在的参数值。LangChain 这类框架已经做了解析兜底但依然不能保证 100% 正确。第二工具本身的边界要设计。比如 TextToSQL 里如果你把“执行 SQL”的能力直接暴露给模型它会尝试查询系统表、查所有表结构、生成超大的 LIMIT 不出来。工具不是“越强大越好”而是“越可控越好”。第三失败后缺少恢复机制。Demo 里模型第一次调用失败简单重试一次大概率就好了。但生产环境里数据库连接超时、上游接口不稳定、模型 API 限流都会导致 Agent 卡死在某个环节。没有超时控制、没有重试策略、没有终止条件Agent 会一直循环直到报 “agent execution terminated due to error”。第四Agent 的链路日志极难追踪。普通接口调试时看一次调用的 input/output 就能定位问题。Agent 调用是一个多轮循环每一轮有用户的输入、模型的中间推理、工具调用结果、最终输出。如果没有统一的 trace 日志出问题时你根本无法判断是哪一轮出了问题。所以我一直认为Agent 项目真正的难点不是在 Notebook 里跑通一个例子而是把一个不可控的模型封装在一个可控的工程框架里。这不正是 Harness 工程在做的事吗下面从环境准备开始我们一步步把 TextToSQL 项目落地。4. 环境准备与前置条件为了让下面的代码可以直接跑通我们把环境限制在一个最通用的最小集Python 3.10 或更高版本LangChain 相关库版本以项目实际安装为准建议使用 V1.0 迁移后的 API一个可用的 OpenAI 兼容接口或本地大模型如 Ollama 部署的模型支持工具调用的都行SQLite 3Python 内置不需要额外安装服务dotenv 读取环境变量。先创建一个项目目录并安装依赖mkdir agent-text2sql cd agent-text2sql python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate创建requirements.txtlangchain langchain-core langgraph python-dotenv安装依赖pip install -r requirements.txt然后在项目根目录创建.env写入模型密钥这里以 OpenAI 兼容接口为例OPENAI_API_KEYsk-xxxxxxxx OPENAI_BASE_URLhttps://your-model-endpoint/v1如果你使用的是本地 Ollama 模型可以不用填 API Key但需要在代码里指定模型名称。下面示例会以 OpenAI 兼容接口为主因为绝大多数云服务和本地网关都支持这一套协议。注意本文演示的 API 调用方式是最通用的langchain-openaiChatOpenAI 封装。具体版本以你安装的依赖为准如果遇到弃用提醒按照官方迁移文档替换导入路径即可。5. TextToSQL 项目完整实现TextToSQL 的目标是用户输入一句自然语言系统返回 SQL 查询结果。看起来很简单但实现时需要拆成多个环节识别数据库结构、写 Prompt、生成 SQL、安全校验、执行查询、解释结果。我们用一个员工数据库示例来跑通闭环。5.1 初始化 SQLite 测试数据库先创建一个执行脚本init_db.py生成测试数据# init_db.py import sqlite3 conn sqlite3.connect(employee.db) cursor conn.cursor() cursor.execute(DROP TABLE IF EXISTS employees) cursor.execute(DROP TABLE IF EXISTS departments) cursor.execute( CREATE TABLE departments ( id INTEGER PRIMARY KEY, name TEXT NOT NULL ) ) cursor.execute( CREATE TABLE employees ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, dept_id INTEGER, salary REAL, join_date TEXT, FOREIGN KEY(dept_id) REFERENCES departments(id) ) ) cursor.execute(INSERT INTO departments (id, name) VALUES (1, 技术部), (2, 市场部), (3, 人事部)) cursor.execute( INSERT INTO employees (id, name, dept_id, salary, join_date) VALUES (1, 张三, 1, 20000, 2020-01-01), (2, 李四, 1, 18000, 2021-02-15), (3, 王五, 2, 15000, 2019-06-01), (4, 赵六, 3, 12000, 2022-09-30), (5, 钱七, 2, 16000, 2020-11-11) ) conn.commit() conn.close() print(数据库初始化完成)运行后得到一个简单的员工库两个表、几条测试数据足够验证 TextToSQL 的完整流程。5.2 用 tool 定义 Agent 可调用的工具Agent 工具设计有一个非常重要的原则不要让模型直接操作原始能力而是给它封装好的、带约束的专用接口。我们定义两个工具get_schema返回数据库表结构供模型生成 SQL 时参考。execute_sql执行模型生成的 SQL但内部做安全检查只允许 SELECT、必须带 LIMIT、禁止多个语句。在项目根目录创建tools.py# tools.py import sqlite3 import re from langchain_core.tools import tool def get_connection(db_path: str employee.db) - sqlite3.Connection: conn sqlite3.connect(db_path) conn.row_factory sqlite3.Row return conn tool def get_schema() - str: 返回 employee.db 的完整表结构包含字段名、类型、主键、外键等信息。 conn get_connection() cursor conn.cursor() cursor.execute( SELECT sql FROM sqlite_master WHERE typetable AND name IN (employees, departments) ) schema \n.join(row[sql] for row in cursor.fetchall()) conn.close() return schema tool def execute_sql(sql: str) - str: 执行一条只读 SQL 查询返回结果集的 JSON 字符串。只允许 SELECT 语句。 sql sql.strip().strip(;) # 安全校验只允许单条 SELECT 语句 if not re.match(r^SELECT\s, sql, re.IGNORECASE): return 错误只允许执行 SELECT 查询 if sql.count(;) 0: return 错误不允许一次执行多条语句 # 强制限制返回行数 if LIMIT not in sql.upper(): sql LIMIT 50 # 禁止读取 sqlite_master 等系统表 if re.search(rsqlite_master|sqlite_temp, sql, re.IGNORECASE): return 错误不允许查询系统表 conn get_connection() try: cursor conn.cursor() cursor.execute(sql) rows cursor.fetchall() conn.close() result [dict(row) for row in rows] return str(result) except Exception as e: conn.close() return f执行错误{e}这段代码里核心不是 SQL 本身而是安全约束。你可能会想LLM 不是已经决定要执行什么了吗为什么要做这么多限制原因是模型生成的 SQL 并不可信。它可能因为 Prompt 含糊而生成多表 JOIN 却忘记 WHERE也可能因为训练数据影响生成一个 DELETE 语句更常见的是生成一个不带 LIMIT 的全表扫描。工具层做校验是 Agent 系统最后一道防线。5.3 用 LangGraph 编排 Agent 循环现在我们来写核心的 Agent 编排逻辑。这里用一个简化但完整的 LangGraph 状态图Agent 节点负责“模型决策”Tools 节点负责“执行工具并返回结果”然后根据模型的输出判断是继续调用工具还是直接回答用户。创建agent_graph.py# agent_graph.py import json from typing import TypedDict, Annotated from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END from tools import get_schema, execute_sql # 1. 定义状态对象 class AgentState(TypedDict): messages: list final_answer: str # 2. 初始化模型并绑定工具 model ChatOpenAI(modelgpt-4o-mini, temperature0) tools [get_schema, execute_sql] model_with_tools model.bind_tools(tools) # 3. 定义 Agent 节点模型决定是否调用工具 def agent_node(state: AgentState): response model_with_tools.invoke(state[messages]) return {messages: [response]} # 4. 定义工具执行节点 def tools_node(state: AgentState): last_message state[messages][-1] tool_messages [] for tool_call in last_message.tool_calls: print(f[调用工具] name{tool_call[name]}, args{tool_call[args]}) tool_map { get_schema: get_schema, execute_sql: execute_sql, } selected_tool tool_map[tool_call[name]] result selected_tool.invoke(tool_call[args]) tool_messages.append( ToolMessage( contentresult, tool_call_idtool_call[id], ) ) return {messages: tool_messages} # 5. 定义路由有工具调用则继续循环没有则结束 def should_continue(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: return tools return end # 6. 构建 LangGraph 图 def build_graph(): workflow StateGraph(AgentState) workflow.add_node(agent, agent_node) workflow.add_node(tools, tools_node) workflow.set_entry_point(agent) workflow.add_conditional_edges( agent, should_continue, { tools: tools, end: END, }, ) workflow.add_edge(tools, agent) return workflow.compile()这个代码结构就是一个最标准的 ReAct Agent 循环用户输入进入agent节点模型决定是回答问题还是调用工具如果要调用工具进入tools节点执行然后把工具结果返回给模型模型基于工具结果继续推理直到不再需要调用工具输出最终回答。5.4 给 Agent 穿上 Harness 外壳直接调用编译后的 Agent 图是可以跑的但它不够健壮。比如API 超时怎么办模型输出非法 JSON 怎么办Agent 无限循环怎么办工具执行抛异常怎么办这些都不应该在业务代码里散落处理而是应该在 Harness 层统一解决。创建harness.py把 Agent 的执行过程封装起一套“安全壳”# harness.py import time import json import logging from typing import Optional from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from agent_graph import build_graph logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) logger logging.getLogger(__name__) class AgentRunResult: def __init__(self, answer: str, trace: list, success: bool): self.answer answer self.trace trace self.success success class AgentHarness: def __init__(self, max_retries: int 3, max_steps: int 10): self.graph build_graph() self.max_retries max_retries self.max_steps max_steps def run(self, user_query: str) - AgentRunResult: messages [HumanMessage(contentuser_query)] trace [] step 0 for attempt in range(self.max_retries): try: config {recursion_limit: self.max_steps} result self.graph.invoke( {messages: messages, final_answer: }, configconfig, ) last_message result[messages][-1] answer ( last_message.content if isinstance(last_message, AIMessage) else str(last_message) ) return AgentRunResult(answeranswer, tracetrace, successTrue) except Exception as e: logger.warning(Agent 执行第 %s 次失败: %s, attempt 1, str(e)) time.sleep(1 * (attempt 1)) if attempt self.max_retries - 1: return AgentRunResult( answerfAgent 执行失败已重试 {self.max_retries} 次: {e}, tracetrace, successFalse, ) return AgentRunResult(answer未知错误, tracetrace, successFalse)Harness 的作用在这里就很明显了重试策略Agent 调用可能因为临时超时失败整体重试一次往往就能恢复步骤上限防止 Agent 陷入“调用工具 → 再调用工具”的死循环统一日志每一步执行过程都能输出 Trace 便于排查。5.5 启动入口与用户交互创建main.py# main.py from harness import AgentHarness def main(): harness AgentHarness(max_retries3, max_steps10) while True: user_input input(请输入你的数据库问题输入 exit 退出).strip() if user_input.lower() in (exit, quit): break result harness.run(user_input) print(\n[回答], result.answer) print( * 50) if __name__ __main__: main()运行项目python init_db.py python main.py6. 运行结果与效果验证按照上面代码完整的流程应该是这样启动main.py后输入一个自然语言问题比如请输入你的数据库问题输入 exit 退出每个部门有多少员工输入后Harness 会启动 LangGraphAgent 先调用get_schema获取表结构再把生成的 SQL 传给execute_sql最后根据查询结果生成最终回答。控制台输出大致如下[调用工具] nameget_schema, args{} [调用工具] nameexecute_sql, args{sql: SELECT d.name, COUNT(e.id) AS employee_count FROM departments d LEFT JOIN employees e ON d.id e.dept_id GROUP BY d.name} [回答] 各部门的员工数量如下 - 技术部2人 - 市场部2人 - 人事部1人再次输入请输入你的数据库问题输入 exit 退出工资最高的员工是谁预期会生成一条带 ORDER BY 和 LIMIT 的 SQL最终回答出“张三”。验证成功与否可以分两个层面看第一个层面看功能是否对。最终回答的数据和你在数据库里执行对应 SQL 得到的结果一致这就是成功了。第二个层面看过程是否可控。观察控制台输出的每一步工具调用。如果模型生成 SQL 后execute_sql返回了安全错误比如包含 DELETE说明你的安全校验生效了。如果 Agent 在步骤上限内完成说明流程终止条件设计合理。更严格一点可以从AgentRunResult.trace里提取全部执行轨迹用于后续断言测试。这也是 Harness 工程比裸调 model 更值钱的地方——它能保证“过程可审计”。7. 常见问题与排查思路实际开发过程中你会发现 TextToSQL 甚至整个 Agent 项目的坑远不止“模型生成 SQL 不准”这么简单。下面是我觉得出现频率最高的一些问题问题现象可能原因排查方式解决方案模型不调用工具直接瞎回答工具描述不清晰模型不知道何时该用工具打印模型原始响应检查 tool_calls 是否为 None重写工具 description明确适用场景生成的 SQL 语法错误数据库方言复杂模型对 DDL 细节理解不足在 Prompt 中补充建表语句和字段注释给模型提供更完整的 schema或加 SQL 语法校验与修正环节工具调用后 Agent 一直循环模型反复调用同一个失败工具没有终止策略打开日志判断循环发生在哪个节点Harness 层加 max_steps工具端对重复调用做去重报 agent execution terminated due to error状态图递归超限或某节点抛出未捕获异常看完整 trace 栈判断是工具节点还是模型节点异常增加异常捕获、重试使用 recursion_limit 控制深度API 超时或限流模型服务端不稳定或请求量超出配额查看上游返回的 HTTP 状态码Harness 层加重试退避考虑请求缓存和降级策略SQL 查询结果过大模型生成的 SQL 没有 LIMIT打印工具入参检查 SQL 文本工具层强制加 LIMIT同时校验返回行数阈值工具层返回字段太多模型无法准确提取查询结果 JSON 过大查看 ToolMessage 内容长度对查询结果做截断、聚合或只返回前 N 行多轮对话时 Agent 丢失上下文状态 messages 未保存每次都是新会话检查传入状态是否包含历史消息维护会话级 messages按窗口策略截断这里最值得重点说的一点是“agent execution terminated due to error” 这类问题本质上不是语言模型出错而是编排层出错了。模型只是返回了一个文本或结构是 LangGraph 在执行时发现状态不满足、节点抛了异常、或者递归深度超限。所以排查时不要先去改 Prompt而是先看日志定位是哪一个 node 抛的异常。工具执行的异常、状态字段赋值异常、工具参数类型校验异常这些在 LangGraph 里都会表现出“执行被终止”的形式。另一个非常常见的坑是模型“认为自己调用了工具”但工具调用实际上失败它下一轮编造了一个结果。这个问题最难查因为最终回答看起来内容合理。唯一可靠的的办法就是把 ToolMessage 的真实返回内容打到日志里人工核对“模型回答中引用的数据”和“工具返回值”是否一致。数据不一致就是模型幻觉。8. 最佳实践与工程建议前面代码已经把最小闭环跑通了。如果要在真实项目里落地还需要补充下面几个层面的实践。8.1 工具设计少而精每个工具只做一件事给 Agent 设计工具时不要想着“一个工具只做查询、一个工具做更新”——太宽泛的工具会被模型滥用。更好的做法是把业务语义封装进工具描述里。例如如果数据库里有很多表不要只给一个execute_sql而是提供query_employee_by_dept(dept_name)get_top_salary_employees(limit)get_department_summary()。每个工具的输入参数有明确的业务含义模型决策时犯错概率会显著下降。工具描述里还要说明“什么时候不能用这个工具”。8.2 SQL 安全永远不要信任模型生成的 SQL前面代码里已经做了基础的安全限制但真实项目还要考虑更多使用只读数据库账号授予的最小权限只包含 SELECT对敏感列做脱敏所有查询强制经过查询分析器禁止子查询嵌套过深使用数据库的 statement_timeout比如 PostgreSQL限制单条 SQL 执行时间生产环境建议把模型生成 SQL 与真正执行 SQL 分离先让模型生成、再让一个规则校验器检查、最后再由人确认关键语句。记住工具层校验不是对模型的不信任而是对自己系统的负责。8.3 Prompt 与 Schema 的细节TextToSQL 项目里模型的 Prompt 通常包含三部分任务描述、表结构、用户问题。表结构描述越清楚SQL 越准。一个常见优化技巧是不要只给字段名和类型还要给字段的业务含义、是否可空、单位、枚举值。例如employees.salary (REAL) - 员工月薪单位元可能为 NULL表示离职。另一个技巧是把“用户问题”做一次改写扩展同义词。例如用户问“薪水”“工资”“薪资”实际上都对应 salary 字段。可以在 Prompt 中加入字段别名映射也可以用一个单独的改写节点先标准化问题再进入 SQL 生成节点。8.4 Agent 失败恢复的层次Harness 层的重试只是最外层真正的生产系统应该有分层次策略模型层当模型返回异常格式时尝试重新解析工具层当工具执行返回错误时把错误信息拼进下一轮模型输入让模型自己修正流程层当某节点重试仍然失败切换到降级方案比如返回“当前无法获取数据请稍后重试”人工层对于关键操作失败后创建工单让运维人员介入。不要把 Agent 的失败恢复全部寄托在“让模型再试一次”上。模型重试的成功率随着尝试次数递减正确做法是把错误信息结构化地反馈给它而不是无脑重新生成。8.5 日志与可观测性Agent 的日志建议至少包含以下字段trace_id一次用户请求的全链路 IDstep_index当前步骤序号node_name当前节点名称input_messages模型输入注意脱敏tool_calls模型请求的工具调用列表tool_results工具实际返回结果latency_ms本步骤耗时error_message异常信息。LangChain 体系里有 LangSmith 等可观测性工具但即使不用商业产品最好也在自己的 Harness 里输出一份完整日志。没有 Trace 的 Agent 项目上线后会非常难维护。8.6 团队协作与代码结构Agent 项目不要把所有代码堆在一个文件里。建议按下面分层组织agent-project/ ├── tools/ # 工具定义 │ ├── __init__.py │ ├── db_tools.py # 数据库工具 │ └── api_tools.py # 上游接口工具 ├── agents/ # Agent 编排 │ ├── graph.py # LangGraph 图定义 │ └── nodes.py # 节点逻辑 ├── harness/ # Harness 层 │ ├── runner.py # 执行与重试 │ ├── validator.py # 输出校验 │ └── logger.py # 日志封装 ├── prompts/ # Prompt 模板独立成文件便于 review │ └── text2sql.py ├── tests/ # 测试用例 └── config.yaml # 模型、工具、Harness 参数配置这个结构本身就是 Harness 工程思想在代码组织上的体现模型 Agent 只做决策业务能力和工程能力都非常清晰地分离开。9. 总结与后续学习方向这篇文章的核心是把大模型 Agent 开发的四个关键认知讲透了第一LangChain V1.0 时代编排层的重点已经从 chain 转向 graph。与其花时间背各种 chain 类不如先掌握 LangGraph 的节点、边、状态化设计。它是 Agent 循环的本质。第二不要裸调大模型要给 Agent 穿 Harness。重试、校验、日志、步骤控制、权限边界这些工程机制才是 Agent 生产可用的关键。第三工具设计决定 Agent 能力上限。TextToSQL 里如何设计工具、如何做安全约束、如何把业务语义塞进工具描述直接决定模型能不能用对。第四TextToSQL 是一个非常适合实战的切入点。它同时涉及结构化数据、模型推理、工具调用、安全控制是一条能完整看到 Agent 开发全貌的“最小价值链”。如果你想继续深入建议按这个顺序去实践把本文的 TextToSQL 示例改造成你熟悉的业务数据库比如电商订单、内容库重点体会 schema 描述对 SQL 准确率的影响在 Harness 层加入输出校验对模型最终答案做关键词或格式校验失败时触发一次“修正对话”用 LangGraph 加入一个人工确认节点当模型想执行 UPDATE/DELETE 时先暂停等用户确认后再执行接入真实生产日志系统把 Agent 的 Trace 变成可分析的数据。大模型 Agent 开发真正拉开水平差距的地方从来不是谁调 API 更熟练而是谁能把不可控的模型放进一个可控的工程系统里。这篇文章给出的代码就是一个能继续生长的起点。建议收藏备用下次写 Agent 项目时可以直接用它当脚手架。 LangChain V1.0与LangGraph下Agent开发：TextToSQL实战与Harness工程
最近半年大模型 Agent 已经从一个偏研究向的概念迅速变成了简历上的关键词、面试必问题和企业立项书里的常客。但真正动手写过 Agent 的开发者大多会有同一个感受Demo 跑起来很快一上生产就各种崩。LangChain 的 API 变来变去、LangGraph 的状态图设计要重新学一遍、Tool 调用时模型经常“假装调用”但输出格式不对、好不容易跑通了结果 SQL 生成出来根本不能执行。这些坑几乎每个 Agent 开发人员都会遇到。而如果你关注行业热点还会看到一个新的高频词——Harness 工程。它到底是什么跟 Agent 开发有什么关系这篇文章不是一份简单的新特性播报。我会用一条完整的TextToSQL 项目落地路径把 LangChain V1.0 的编排思路、LangGraph 的状态流设计、Agent 工具链的写法、以及 Harness 工程对生产可用性的价值串起来讲清楚。读完你会明白为什么 Agent 项目不能只靠“堆 Prompt 调 API”以及一个能稳定运行的智能体系统工程结构到底长什么样。1. 这篇文章真正要解决的问题先给一个明确判断当前大模型 Agent 开发的瓶颈不再是模型能力而是工程能力。模型能理解多少、能生成多好已经高度收敛但“能不能稳定执行”“失败后能不能自愈”“工具调用错不乱来”这些工程问题才是拉开项目差距的地方。如果你正在做以下事情这篇文章就是写给你的你已经在调大模型 API想从“单轮问答”升级到“能调用工具的 Agent”你被 LangChain 老版本的一堆 chain API 弄晕了想搞清楚 V1.0 到底该学什么你想做 TextToSQL / NL2SQL 项目但发现模型生成的 SQL 经常语法错、执行错、甚至查出和问题无关的数据你听说了 Harness 工程但不知道它在大模型项目里到底落在哪一层你的 Agent 经常出现 “agent execution terminated due to error” 这类问题想知道从哪里排查。这篇文章会以 TextToSQL 为例横跨四个层级来讲模型层怎么选与调、编排层用什么框架、工具层怎么设计、工程层怎么保证稳定。最终代码会跑通一个完整的“自然语言查数据库”最小闭环。2. 核心概念LangChain、LangGraph、Agent 与 Harness 工程很多初学者会把 LangChain 和 LangGraph 混为一谈这是第一个要纠正的认知。LangChain 是一个大模型应用编排框架。它的核心价值是帮你统一管理大模型调用、Prompt 模板、输出解析、外部工具调用等一堆琐碎的事情。你可以在 LangChain 里不用自己写繁琐的 HTTP 封装和 JSON 解析直接定义模型、工具和流程。不过LangChain 早期的 Chain API 抽象层次偏重很多人学过之后真正写项目时反而觉得不如直接用原生代码舒服。LangGraph 是面向“有状态、多步骤、有条件判断”的 Agent 流程编排框架。你可以把它理解成一个“把 Agent 流程画成流程图”的开发库。Node 是流程中的每一个步骤Edge 是步骤之间的跳转条件State 是在整个流程中流动的数据对象。LangGraph 最大的价值在于Agent 的执行不再是“一次调用就结束”而是可以循环模型决定调用工具 → 工具返回结果 → 模型继续分析 → 再决定调用下一个工具。这种循环在原生代码里写起来很容易乱用 LangGraph 则清晰很多。Agent 是指能自主规划任务、调用外部工具、并根据工具结果继续行动的智能体程序。跟普通大模型应用的关键区别在于普通应用是“用户问一句模型答一句”Agent 则是一个循环系统它有感知读取工具输入、决策判断下一步做什么、行动调用工具、观察读取工具结果四个阶段。Harness 工程是大模型 Agent 开发中一个很新的工程化概念。英文 “test harness” 在软件工程里指测试夹具/测试框架而在大模型领域Harness 工程指的是“把模型能力包裹进一套可控制、可验证、可重试、可观测的运行框架中”。说得直白一点Harness 就是 Agent 的执行安全壳。它负责管理请求重试、失败降级、输出校验、权限控制、日志追踪等与模型能力无关、但决定系统能否上线的脏活累活。传统应用开发里你从来不会让用户直接执行一段不可控的 SQL同样在你写 Agent 时你也不应该让模型直接裸调数据库、裸发 HTTP 请求。Harness 工程要解决的就是这个“接口可信”的问题。下面用一个表格对比它们的定位概念定位类比关键问题LangChain基础编排框架工具装箱统一 API怎么管理 Model / Prompt / ToolLangGraph流程状态编排流程图 状态机Agent 多步执行时如何跳转Agent智能体应用范式会思考的行动体怎么决策、怎么调用工具Harness 工程生产级运行外壳测试框架 安全壳失败重试、输出校验、日志、权限注意这四个概念不是并列的技术而是不同层的产物。LangChain 和 LangGraph 是“底座”Agent 是“模式”Harness 工程是“生产化的封装方式”。这也解释了为什么 LangChain 官方在 V1.0 之后重点推 LangGraph——因为单纯的“链式串联”已经满足不了 Agent 的复杂流程控制需求。3. Agent 开发为什么难从 Demo 到可用的鸿沟很多人写 Agent 的起点是照着一个 LangChain 官方示例改出来的。示例里有一个 ReAct Agent模型自己能看到工具描述、自己决定是否调用跑一个简单计算题或搜索题看起来真的“智能”。于是你想那我把数据库查询、接口调用都封装成工具模型不就能帮我干活了吗真实情况是一旦把 Agent 接入真实业务问题立刻涌出来第一模型输出不稳定。模型“决定调用工具”的方式是输出一个结构化的工具调用对象。但真实项目中你可能会遇到返回了空字符串、返回了拼写错误的方法名、返回了根本不存在的参数值。LangChain 这类框架已经做了解析兜底但依然不能保证 100% 正确。第二工具本身的边界要设计。比如 TextToSQL 里如果你把“执行 SQL”的能力直接暴露给模型它会尝试查询系统表、查所有表结构、生成超大的 LIMIT 不出来。工具不是“越强大越好”而是“越可控越好”。第三失败后缺少恢复机制。Demo 里模型第一次调用失败简单重试一次大概率就好了。但生产环境里数据库连接超时、上游接口不稳定、模型 API 限流都会导致 Agent 卡死在某个环节。没有超时控制、没有重试策略、没有终止条件Agent 会一直循环直到报 “agent execution terminated due to error”。第四Agent 的链路日志极难追踪。普通接口调试时看一次调用的 input/output 就能定位问题。Agent 调用是一个多轮循环每一轮有用户的输入、模型的中间推理、工具调用结果、最终输出。如果没有统一的 trace 日志出问题时你根本无法判断是哪一轮出了问题。所以我一直认为Agent 项目真正的难点不是在 Notebook 里跑通一个例子而是把一个不可控的模型封装在一个可控的工程框架里。这不正是 Harness 工程在做的事吗下面从环境准备开始我们一步步把 TextToSQL 项目落地。4. 环境准备与前置条件为了让下面的代码可以直接跑通我们把环境限制在一个最通用的最小集Python 3.10 或更高版本LangChain 相关库版本以项目实际安装为准建议使用 V1.0 迁移后的 API一个可用的 OpenAI 兼容接口或本地大模型如 Ollama 部署的模型支持工具调用的都行SQLite 3Python 内置不需要额外安装服务dotenv 读取环境变量。先创建一个项目目录并安装依赖mkdir agent-text2sql cd agent-text2sql python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate创建requirements.txtlangchain langchain-core langgraph python-dotenv安装依赖pip install -r requirements.txt然后在项目根目录创建.env写入模型密钥这里以 OpenAI 兼容接口为例OPENAI_API_KEYsk-xxxxxxxx OPENAI_BASE_URLhttps://your-model-endpoint/v1如果你使用的是本地 Ollama 模型可以不用填 API Key但需要在代码里指定模型名称。下面示例会以 OpenAI 兼容接口为主因为绝大多数云服务和本地网关都支持这一套协议。注意本文演示的 API 调用方式是最通用的langchain-openaiChatOpenAI 封装。具体版本以你安装的依赖为准如果遇到弃用提醒按照官方迁移文档替换导入路径即可。5. TextToSQL 项目完整实现TextToSQL 的目标是用户输入一句自然语言系统返回 SQL 查询结果。看起来很简单但实现时需要拆成多个环节识别数据库结构、写 Prompt、生成 SQL、安全校验、执行查询、解释结果。我们用一个员工数据库示例来跑通闭环。5.1 初始化 SQLite 测试数据库先创建一个执行脚本init_db.py生成测试数据# init_db.py import sqlite3 conn sqlite3.connect(employee.db) cursor conn.cursor() cursor.execute(DROP TABLE IF EXISTS employees) cursor.execute(DROP TABLE IF EXISTS departments) cursor.execute( CREATE TABLE departments ( id INTEGER PRIMARY KEY, name TEXT NOT NULL ) ) cursor.execute( CREATE TABLE employees ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, dept_id INTEGER, salary REAL, join_date TEXT, FOREIGN KEY(dept_id) REFERENCES departments(id) ) ) cursor.execute(INSERT INTO departments (id, name) VALUES (1, 技术部), (2, 市场部), (3, 人事部)) cursor.execute( INSERT INTO employees (id, name, dept_id, salary, join_date) VALUES (1, 张三, 1, 20000, 2020-01-01), (2, 李四, 1, 18000, 2021-02-15), (3, 王五, 2, 15000, 2019-06-01), (4, 赵六, 3, 12000, 2022-09-30), (5, 钱七, 2, 16000, 2020-11-11) ) conn.commit() conn.close() print(数据库初始化完成)运行后得到一个简单的员工库两个表、几条测试数据足够验证 TextToSQL 的完整流程。5.2 用 tool 定义 Agent 可调用的工具Agent 工具设计有一个非常重要的原则不要让模型直接操作原始能力而是给它封装好的、带约束的专用接口。我们定义两个工具get_schema返回数据库表结构供模型生成 SQL 时参考。execute_sql执行模型生成的 SQL但内部做安全检查只允许 SELECT、必须带 LIMIT、禁止多个语句。在项目根目录创建tools.py# tools.py import sqlite3 import re from langchain_core.tools import tool def get_connection(db_path: str employee.db) - sqlite3.Connection: conn sqlite3.connect(db_path) conn.row_factory sqlite3.Row return conn tool def get_schema() - str: 返回 employee.db 的完整表结构包含字段名、类型、主键、外键等信息。 conn get_connection() cursor conn.cursor() cursor.execute( SELECT sql FROM sqlite_master WHERE typetable AND name IN (employees, departments) ) schema \n.join(row[sql] for row in cursor.fetchall()) conn.close() return schema tool def execute_sql(sql: str) - str: 执行一条只读 SQL 查询返回结果集的 JSON 字符串。只允许 SELECT 语句。 sql sql.strip().strip(;) # 安全校验只允许单条 SELECT 语句 if not re.match(r^SELECT\s, sql, re.IGNORECASE): return 错误只允许执行 SELECT 查询 if sql.count(;) 0: return 错误不允许一次执行多条语句 # 强制限制返回行数 if LIMIT not in sql.upper(): sql LIMIT 50 # 禁止读取 sqlite_master 等系统表 if re.search(rsqlite_master|sqlite_temp, sql, re.IGNORECASE): return 错误不允许查询系统表 conn get_connection() try: cursor conn.cursor() cursor.execute(sql) rows cursor.fetchall() conn.close() result [dict(row) for row in rows] return str(result) except Exception as e: conn.close() return f执行错误{e}这段代码里核心不是 SQL 本身而是安全约束。你可能会想LLM 不是已经决定要执行什么了吗为什么要做这么多限制原因是模型生成的 SQL 并不可信。它可能因为 Prompt 含糊而生成多表 JOIN 却忘记 WHERE也可能因为训练数据影响生成一个 DELETE 语句更常见的是生成一个不带 LIMIT 的全表扫描。工具层做校验是 Agent 系统最后一道防线。5.3 用 LangGraph 编排 Agent 循环现在我们来写核心的 Agent 编排逻辑。这里用一个简化但完整的 LangGraph 状态图Agent 节点负责“模型决策”Tools 节点负责“执行工具并返回结果”然后根据模型的输出判断是继续调用工具还是直接回答用户。创建agent_graph.py# agent_graph.py import json from typing import TypedDict, Annotated from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END from tools import get_schema, execute_sql # 1. 定义状态对象 class AgentState(TypedDict): messages: list final_answer: str # 2. 初始化模型并绑定工具 model ChatOpenAI(modelgpt-4o-mini, temperature0) tools [get_schema, execute_sql] model_with_tools model.bind_tools(tools) # 3. 定义 Agent 节点模型决定是否调用工具 def agent_node(state: AgentState): response model_with_tools.invoke(state[messages]) return {messages: [response]} # 4. 定义工具执行节点 def tools_node(state: AgentState): last_message state[messages][-1] tool_messages [] for tool_call in last_message.tool_calls: print(f[调用工具] name{tool_call[name]}, args{tool_call[args]}) tool_map { get_schema: get_schema, execute_sql: execute_sql, } selected_tool tool_map[tool_call[name]] result selected_tool.invoke(tool_call[args]) tool_messages.append( ToolMessage( contentresult, tool_call_idtool_call[id], ) ) return {messages: tool_messages} # 5. 定义路由有工具调用则继续循环没有则结束 def should_continue(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: return tools return end # 6. 构建 LangGraph 图 def build_graph(): workflow StateGraph(AgentState) workflow.add_node(agent, agent_node) workflow.add_node(tools, tools_node) workflow.set_entry_point(agent) workflow.add_conditional_edges( agent, should_continue, { tools: tools, end: END, }, ) workflow.add_edge(tools, agent) return workflow.compile()这个代码结构就是一个最标准的 ReAct Agent 循环用户输入进入agent节点模型决定是回答问题还是调用工具如果要调用工具进入tools节点执行然后把工具结果返回给模型模型基于工具结果继续推理直到不再需要调用工具输出最终回答。5.4 给 Agent 穿上 Harness 外壳直接调用编译后的 Agent 图是可以跑的但它不够健壮。比如API 超时怎么办模型输出非法 JSON 怎么办Agent 无限循环怎么办工具执行抛异常怎么办这些都不应该在业务代码里散落处理而是应该在 Harness 层统一解决。创建harness.py把 Agent 的执行过程封装起一套“安全壳”# harness.py import time import json import logging from typing import Optional from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from agent_graph import build_graph logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) logger logging.getLogger(__name__) class AgentRunResult: def __init__(self, answer: str, trace: list, success: bool): self.answer answer self.trace trace self.success success class AgentHarness: def __init__(self, max_retries: int 3, max_steps: int 10): self.graph build_graph() self.max_retries max_retries self.max_steps max_steps def run(self, user_query: str) - AgentRunResult: messages [HumanMessage(contentuser_query)] trace [] step 0 for attempt in range(self.max_retries): try: config {recursion_limit: self.max_steps} result self.graph.invoke( {messages: messages, final_answer: }, configconfig, ) last_message result[messages][-1] answer ( last_message.content if isinstance(last_message, AIMessage) else str(last_message) ) return AgentRunResult(answeranswer, tracetrace, successTrue) except Exception as e: logger.warning(Agent 执行第 %s 次失败: %s, attempt 1, str(e)) time.sleep(1 * (attempt 1)) if attempt self.max_retries - 1: return AgentRunResult( answerfAgent 执行失败已重试 {self.max_retries} 次: {e}, tracetrace, successFalse, ) return AgentRunResult(answer未知错误, tracetrace, successFalse)Harness 的作用在这里就很明显了重试策略Agent 调用可能因为临时超时失败整体重试一次往往就能恢复步骤上限防止 Agent 陷入“调用工具 → 再调用工具”的死循环统一日志每一步执行过程都能输出 Trace 便于排查。5.5 启动入口与用户交互创建main.py# main.py from harness import AgentHarness def main(): harness AgentHarness(max_retries3, max_steps10) while True: user_input input(请输入你的数据库问题输入 exit 退出).strip() if user_input.lower() in (exit, quit): break result harness.run(user_input) print(\n[回答], result.answer) print( * 50) if __name__ __main__: main()运行项目python init_db.py python main.py6. 运行结果与效果验证按照上面代码完整的流程应该是这样启动main.py后输入一个自然语言问题比如请输入你的数据库问题输入 exit 退出每个部门有多少员工输入后Harness 会启动 LangGraphAgent 先调用get_schema获取表结构再把生成的 SQL 传给execute_sql最后根据查询结果生成最终回答。控制台输出大致如下[调用工具] nameget_schema, args{} [调用工具] nameexecute_sql, args{sql: SELECT d.name, COUNT(e.id) AS employee_count FROM departments d LEFT JOIN employees e ON d.id e.dept_id GROUP BY d.name} [回答] 各部门的员工数量如下 - 技术部2人 - 市场部2人 - 人事部1人再次输入请输入你的数据库问题输入 exit 退出工资最高的员工是谁预期会生成一条带 ORDER BY 和 LIMIT 的 SQL最终回答出“张三”。验证成功与否可以分两个层面看第一个层面看功能是否对。最终回答的数据和你在数据库里执行对应 SQL 得到的结果一致这就是成功了。第二个层面看过程是否可控。观察控制台输出的每一步工具调用。如果模型生成 SQL 后execute_sql返回了安全错误比如包含 DELETE说明你的安全校验生效了。如果 Agent 在步骤上限内完成说明流程终止条件设计合理。更严格一点可以从AgentRunResult.trace里提取全部执行轨迹用于后续断言测试。这也是 Harness 工程比裸调 model 更值钱的地方——它能保证“过程可审计”。7. 常见问题与排查思路实际开发过程中你会发现 TextToSQL 甚至整个 Agent 项目的坑远不止“模型生成 SQL 不准”这么简单。下面是我觉得出现频率最高的一些问题问题现象可能原因排查方式解决方案模型不调用工具直接瞎回答工具描述不清晰模型不知道何时该用工具打印模型原始响应检查 tool_calls 是否为 None重写工具 description明确适用场景生成的 SQL 语法错误数据库方言复杂模型对 DDL 细节理解不足在 Prompt 中补充建表语句和字段注释给模型提供更完整的 schema或加 SQL 语法校验与修正环节工具调用后 Agent 一直循环模型反复调用同一个失败工具没有终止策略打开日志判断循环发生在哪个节点Harness 层加 max_steps工具端对重复调用做去重报 agent execution terminated due to error状态图递归超限或某节点抛出未捕获异常看完整 trace 栈判断是工具节点还是模型节点异常增加异常捕获、重试使用 recursion_limit 控制深度API 超时或限流模型服务端不稳定或请求量超出配额查看上游返回的 HTTP 状态码Harness 层加重试退避考虑请求缓存和降级策略SQL 查询结果过大模型生成的 SQL 没有 LIMIT打印工具入参检查 SQL 文本工具层强制加 LIMIT同时校验返回行数阈值工具层返回字段太多模型无法准确提取查询结果 JSON 过大查看 ToolMessage 内容长度对查询结果做截断、聚合或只返回前 N 行多轮对话时 Agent 丢失上下文状态 messages 未保存每次都是新会话检查传入状态是否包含历史消息维护会话级 messages按窗口策略截断这里最值得重点说的一点是“agent execution terminated due to error” 这类问题本质上不是语言模型出错而是编排层出错了。模型只是返回了一个文本或结构是 LangGraph 在执行时发现状态不满足、节点抛了异常、或者递归深度超限。所以排查时不要先去改 Prompt而是先看日志定位是哪一个 node 抛的异常。工具执行的异常、状态字段赋值异常、工具参数类型校验异常这些在 LangGraph 里都会表现出“执行被终止”的形式。另一个非常常见的坑是模型“认为自己调用了工具”但工具调用实际上失败它下一轮编造了一个结果。这个问题最难查因为最终回答看起来内容合理。唯一可靠的的办法就是把 ToolMessage 的真实返回内容打到日志里人工核对“模型回答中引用的数据”和“工具返回值”是否一致。数据不一致就是模型幻觉。8. 最佳实践与工程建议前面代码已经把最小闭环跑通了。如果要在真实项目里落地还需要补充下面几个层面的实践。8.1 工具设计少而精每个工具只做一件事给 Agent 设计工具时不要想着“一个工具只做查询、一个工具做更新”——太宽泛的工具会被模型滥用。更好的做法是把业务语义封装进工具描述里。例如如果数据库里有很多表不要只给一个execute_sql而是提供query_employee_by_dept(dept_name)get_top_salary_employees(limit)get_department_summary()。每个工具的输入参数有明确的业务含义模型决策时犯错概率会显著下降。工具描述里还要说明“什么时候不能用这个工具”。8.2 SQL 安全永远不要信任模型生成的 SQL前面代码里已经做了基础的安全限制但真实项目还要考虑更多使用只读数据库账号授予的最小权限只包含 SELECT对敏感列做脱敏所有查询强制经过查询分析器禁止子查询嵌套过深使用数据库的 statement_timeout比如 PostgreSQL限制单条 SQL 执行时间生产环境建议把模型生成 SQL 与真正执行 SQL 分离先让模型生成、再让一个规则校验器检查、最后再由人确认关键语句。记住工具层校验不是对模型的不信任而是对自己系统的负责。8.3 Prompt 与 Schema 的细节TextToSQL 项目里模型的 Prompt 通常包含三部分任务描述、表结构、用户问题。表结构描述越清楚SQL 越准。一个常见优化技巧是不要只给字段名和类型还要给字段的业务含义、是否可空、单位、枚举值。例如employees.salary (REAL) - 员工月薪单位元可能为 NULL表示离职。另一个技巧是把“用户问题”做一次改写扩展同义词。例如用户问“薪水”“工资”“薪资”实际上都对应 salary 字段。可以在 Prompt 中加入字段别名映射也可以用一个单独的改写节点先标准化问题再进入 SQL 生成节点。8.4 Agent 失败恢复的层次Harness 层的重试只是最外层真正的生产系统应该有分层次策略模型层当模型返回异常格式时尝试重新解析工具层当工具执行返回错误时把错误信息拼进下一轮模型输入让模型自己修正流程层当某节点重试仍然失败切换到降级方案比如返回“当前无法获取数据请稍后重试”人工层对于关键操作失败后创建工单让运维人员介入。不要把 Agent 的失败恢复全部寄托在“让模型再试一次”上。模型重试的成功率随着尝试次数递减正确做法是把错误信息结构化地反馈给它而不是无脑重新生成。8.5 日志与可观测性Agent 的日志建议至少包含以下字段trace_id一次用户请求的全链路 IDstep_index当前步骤序号node_name当前节点名称input_messages模型输入注意脱敏tool_calls模型请求的工具调用列表tool_results工具实际返回结果latency_ms本步骤耗时error_message异常信息。LangChain 体系里有 LangSmith 等可观测性工具但即使不用商业产品最好也在自己的 Harness 里输出一份完整日志。没有 Trace 的 Agent 项目上线后会非常难维护。8.6 团队协作与代码结构Agent 项目不要把所有代码堆在一个文件里。建议按下面分层组织agent-project/ ├── tools/ # 工具定义 │ ├── __init__.py │ ├── db_tools.py # 数据库工具 │ └── api_tools.py # 上游接口工具 ├── agents/ # Agent 编排 │ ├── graph.py # LangGraph 图定义 │ └── nodes.py # 节点逻辑 ├── harness/ # Harness 层 │ ├── runner.py # 执行与重试 │ ├── validator.py # 输出校验 │ └── logger.py # 日志封装 ├── prompts/ # Prompt 模板独立成文件便于 review │ └── text2sql.py ├── tests/ # 测试用例 └── config.yaml # 模型、工具、Harness 参数配置这个结构本身就是 Harness 工程思想在代码组织上的体现模型 Agent 只做决策业务能力和工程能力都非常清晰地分离开。9. 总结与后续学习方向这篇文章的核心是把大模型 Agent 开发的四个关键认知讲透了第一LangChain V1.0 时代编排层的重点已经从 chain 转向 graph。与其花时间背各种 chain 类不如先掌握 LangGraph 的节点、边、状态化设计。它是 Agent 循环的本质。第二不要裸调大模型要给 Agent 穿 Harness。重试、校验、日志、步骤控制、权限边界这些工程机制才是 Agent 生产可用的关键。第三工具设计决定 Agent 能力上限。TextToSQL 里如何设计工具、如何做安全约束、如何把业务语义塞进工具描述直接决定模型能不能用对。第四TextToSQL 是一个非常适合实战的切入点。它同时涉及结构化数据、模型推理、工具调用、安全控制是一条能完整看到 Agent 开发全貌的“最小价值链”。如果你想继续深入建议按这个顺序去实践把本文的 TextToSQL 示例改造成你熟悉的业务数据库比如电商订单、内容库重点体会 schema 描述对 SQL 准确率的影响在 Harness 层加入输出校验对模型最终答案做关键词或格式校验失败时触发一次“修正对话”用 LangGraph 加入一个人工确认节点当模型想执行 UPDATE/DELETE 时先暂停等用户确认后再执行接入真实生产日志系统把 Agent 的 Trace 变成可分析的数据。大模型 Agent 开发真正拉开水平差距的地方从来不是谁调 API 更熟练而是谁能把不可控的模型放进一个可控的工程系统里。这篇文章给出的代码就是一个能继续生长的起点。建议收藏备用下次写 Agent 项目时可以直接用它当脚手架。