LangChain V1.0与LangGraph下Agent开发:TextToSQL实战与Harness工程
2026/9/7 3:55:10 网站建设 项目流程

最近半年,大模型 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 / Tool
LangGraph流程状态编排流程图 + 状态机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 3(Python 内置,不需要额外安装服务);
  • dotenv 读取环境变量。

先创建一个项目目录并安装依赖:

mkdir agent-text2sql cd agent-text2sql python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate

创建requirements.txt

langchain langchain-core langgraph python-dotenv

安装依赖:

pip install -r requirements.txt

然后在项目根目录创建.env,写入模型密钥(这里以 OpenAI 兼容接口为例):

OPENAI_API_KEY=sk-xxxxxxxx OPENAI_BASE_URL=https://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 工具设计有一个非常重要的原则:不要让模型直接操作原始能力,而是给它封装好的、带约束的专用接口。

我们定义两个工具:

  1. get_schema:返回数据库表结构,供模型生成 SQL 时参考。
  2. 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 type='table' 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(r"sqlite_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(model="gpt-4o-mini", temperature=0) 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( content=result, tool_call_id=tool_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 循环:

  1. 用户输入进入agent节点;
  2. 模型决定是回答问题,还是调用工具;
  3. 如果要调用工具,进入tools节点执行,然后把工具结果返回给模型;
  4. 模型基于工具结果继续推理,直到不再需要调用工具,输出最终回答。

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(level=logging.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(content=user_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": ""}, config=config, ) last_message = result["messages"][-1] answer = ( last_message.content if isinstance(last_message, AIMessage) else str(last_message) ) return AgentRunResult(answer=answer, trace=trace, success=True) 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( answer=f"Agent 执行失败,已重试 {self.max_retries} 次: {e}", trace=trace, success=False, ) return AgentRunResult(answer="未知错误", trace=trace, success=False)

Harness 的作用在这里就很明显了:

  • 重试策略:Agent 调用可能因为临时超时失败,整体重试一次往往就能恢复;
  • 步骤上限:防止 Agent 陷入“调用工具 → 再调用工具”的死循环;
  • 统一日志:每一步执行过程都能输出 Trace 便于排查。

5.5 启动入口与用户交互

创建main.py

# main.py from harness import AgentHarness def main(): harness = AgentHarness(max_retries=3, max_steps=10) 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.py

6. 运行结果与效果验证

按照上面代码,完整的流程应该是这样:

启动main.py后,输入一个自然语言问题,比如:

请输入你的数据库问题(输入 exit 退出):每个部门有多少员工

输入后,Harness 会启动 LangGraph,Agent 先调用get_schema获取表结构,再把生成的 SQL 传给execute_sql,最后根据查询结果生成最终回答。

控制台输出大致如下:

[调用工具] name=get_schema, args={} [调用工具] name=execute_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 层的重试只是最外层,真正的生产系统应该有分层次策略:

  1. 模型层:当模型返回异常格式时,尝试重新解析;
  2. 工具层:当工具执行返回错误时,把错误信息拼进下一轮模型输入,让模型自己修正;
  3. 流程层:当某节点重试仍然失败,切换到降级方案,比如返回“当前无法获取数据,请稍后重试”;
  4. 人工层:对于关键操作,失败后创建工单让运维人员介入。

不要把 Agent 的失败恢复全部寄托在“让模型再试一次”上。模型重试的成功率随着尝试次数递减,正确做法是把错误信息结构化地反馈给它,而不是无脑重新生成。

8.5 日志与可观测性

Agent 的日志,建议至少包含以下字段:

  • trace_id:一次用户请求的全链路 ID;
  • step_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 开发全貌的“最小价值链”。

如果你想继续深入,建议按这个顺序去实践:

  1. 把本文的 TextToSQL 示例改造成你熟悉的业务数据库(比如电商订单、内容库),重点体会 schema 描述对 SQL 准确率的影响;
  2. 在 Harness 层加入输出校验:对模型最终答案做关键词或格式校验,失败时触发一次“修正对话”;
  3. 用 LangGraph 加入一个人工确认节点:当模型想执行 UPDATE/DELETE 时,先暂停,等用户确认后再执行;
  4. 接入真实生产日志系统,把 Agent 的 Trace 变成可分析的数据。

大模型 Agent 开发,真正拉开水平差距的地方,从来不是谁调 API 更熟练,而是谁能把不可控的模型,放进一个可控的工程系统里。这篇文章给出的代码,就是一个能继续生长的起点。建议收藏备用,下次写 Agent 项目时,可以直接用它当脚手架。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询