最近在做一个内部报表自动化的项目目标是让业务同学用自然语言直接查数。比如输入“上个月华东区销售额最高的商品Top 10”系统自动生成 SQL、查出结果、再返回一句人话。听起来不算复杂真正动手才发现里面全是细节模型选型、工具定义、多轮对话状态、SQL 安全管控、可观测性……网上资料很多但大多是孤立片段今天把整条链路完整整理出来。这篇文章会围绕大模型 Agent 开发展开内容偏全栈落地。你会看到 LangChain 与 LangGraph 的关系与分工、Harness 工程在 LLM 应用中的定位、智能体工具Tool Calling的完整写法以及一个可直接运行的 TextToSQL 项目实战。无论是刚开始学 Agent 的初学者还是想把它落地到业务系统的开发者都应该能从里面找到可以复用的内容。1. 大模型 Agent 是什么先理清概念再动手1.1 从 LLM 到 Agent为什么需要智能体大模型本身是一个“文本生成器”它的强项是理解语义、生成内容、做推理但它有几个很明显的问题不知道实时数据、不能主动调用外部系统、不能自己执行动作。比如你问模型“帮我把订单表按时间排序导出”它可能能写出 SQL但它没办法真正连接数据库去执行也没办法处理执行后的异常。Agent 的概念就是在这条短板上长出来的。一个完整的 Agent 通常由四部分组成大模型负责理解用户意图、规划步骤、决定调用哪个工具。工具Tools模型可以调用的外部函数比如查数据库、调 API、发邮件、执行计算。执行循环Loop模型输出意图 - 执行工具 - 把结果反馈给模型 - 继续决策。状态管理State记录多轮对话、中间结果、错误信息保证 Agent 不“失忆”。简单说大模型是“大脑”工具是“手脚”Agent 就是把两者串联起来的执行系统。这也是为什么 2024 年之后Agent 几乎成了大模型应用开发的核心关键词。1.2 LangChain 与 LangGraph 的区别和关系很多初学者会混淆 LangChain 和 LangGraph其实它们的定位不一样框架定位侧重点LangChainLLM 应用开发框架提供模型封装、Prompt 模板、工具标准接口、向量库集成等基础组件LangGraphAgent 编排引擎基于图结构管理状态流转适合实现复杂的多轮 Agent 循环可以这么理解LangChain 是“零件库”它帮你把模型、Prompt、工具、记忆这些零件标准化LangGraph 是“流水线图纸”它定义 Agent 每一步怎么走、遇到工具结果后怎么回到模型、什么时候终止。LangChain 进入 V1.0 之后API 设计稳定了很多核心组件从langchain-core中暴露工具定义、模型调用、JSON 结构化输出等都有统一的规范。而 LangGraph 在 1.0 中也成为官方推荐的 Agent 编排方式两者并不是替代关系而是递进关系。在实际项目中我通常是 LangChain 负责模型和工具封装LangGraph 负责把 Agent 流程串起来。1.3 Harness 工程LLM 应用交付的“脚手架”Harness 这个词最近在 Agent 领域出现得很频繁尤其是和 Codex Harness、Agent Harness 相关的技术讨论。它直译是“马具”在软件工程里可以理解为“支撑系统运行的工程外壳”。放到大模型应用场景下Harness 工程指的是在模型能力之上把上下文管理、工具注册、安全策略、日志追踪、配置管理、错误重试等能力封装成一套可复用的运行时环境。它不直接决定模型的“智商”但决定了 Agent 能不能稳定、安全、可观测地跑到生产环境。Harness 和 Agent 的区别可以这样看Agent 是业务逻辑它解决“做什么”Harness 是运行环境它解决“怎么稳定地做”。如果只写一个 Agent 脚本自娱自乐可以不关心 Harness但一旦要上线给业务团队使用就必须考虑工具权限怎么限、Prompt 注入怎么防、执行日志怎么追踪、超时和失败怎么处理这些都属于 Harness 工程要解决的问题。1.4 TextToSQL 为什么是 Agent 项目的最佳练手场景TextToSQL 指把自然语言转换为 SQL 查询语句然后执行并返回结果。它非常适合作为 Agent 实战项目原因有三个核心链路完整涉及意图识别、SQL 生成、SQL 校验、执行、结果解释覆盖 Agent 的完整闭环。边界清晰数据库 schema 是确定的查询结果是可验证的便于评估模型效果。业务价值直接报表、数据分析、运营取数都有刚需做好一个 TextToSQL Agent可以直接搬到很多企业内部系统。当然TextToSQL 也有一堆坑模型生成的 SQL 可能语法错误、可能查全表、可能有权限越界问题。这些正好能引出工具设计、安全策略、错误恢复等进阶话题。2. 环境准备与版本说明在动手之前先把运行环境准备好。大模型 Agent 开发目前还是以 Python 生态最方便下面的示例也全部使用 Python。2.1 编程环境操作系统Windows / macOS / Linux 均可文中命令以 Linux / macOS 风格演示。Python建议 3.10 及以上版本。包管理工具推荐uv或pip示例中使用pip。IDEVS Code 或 PyCharm 都可以。数据库SQLite零配置、跨平台适合本地演示。版本方面说明一下LangChain / LangGraph 迭代速度比较快不同版本的 API 可能有差异本文示例以比较常见的新版 API 风格编写如果你安装的版本更旧需要根据官方迁移文档调整。2.2 安装依赖# 建议先创建虚拟环境 python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate pip install langchain langchain-openai langgraph langchain-community pip install python-dotenv pandas核心依赖说明langchain基础框架。langchain-openaiOpenAI 兼容接口的模型封装。注意很多国产模型也提供 OpenAI 兼容接口所以这个包也能用来对接非 OpenAI 模型。langgraphAgent 图编排框架。langchain-community社区集成某些情况下会用到。python-dotenv读取.env配置文件。2.3 模型 API 准备TextToSQL 对模型的 SQL 生成能力要求较高推荐使用指令遵循能力较强的模型。演示代码中通过环境变量配置模型连接信息不写死具体厂商# .env 文件 LLM_API_KEYsk-xxxx LLM_BASE_URLhttps://api.openai.com/v1 LLM_MODELgpt-4o-mini如果本地有 GPU也可以用 Ollama 部署开源模型然后通过base_url指向本地服务比如LLM_BASE_URLhttp://localhost:11434/v1 LLM_MODELqwen2.5:7b这样做的目的是让代码与具体模型厂商解耦。在企业项目中模型供应商随时可能调整统一走 OpenAI 兼容协议后替换成本会低很多。3. 核心原理拆解Agent 的四个关键部件开始写完整项目之前先拆解 Agent 运行中最重要的几个机制。3.1 模型调用与结构化输出Agent 和普通 LLM 应用最大的区别是模型的输出不只是给人看的文本还要能被程序解析。所以需要让模型输出结构化内容最常见的是 JSON。LangChain 中可以用with_structured_output来要求模型按 Pydantic 模型输出from pydantic import BaseModel, Field from langchain_openai import ChatOpenAI class SQLResult(BaseModel): thought: str Field(description你的思考过程) sql: str Field(description生成的SQL语句) reason: str Field(description为什么这样写) llm ChatOpenAI( modelgpt-4o-mini, api_keysk-xxx, base_urlhttps://api.openai.com/v1, temperature0, ) structured_llm llm.with_structured_output(SQLResult) result structured_llm.invoke(查询销售额最高的五个商品) print(result.sql)这里有几个细节值得注意temperature0TextToSQL 任务是确定性问题温度越低越稳定。with_structured_output不同的模型实现机制不同有的走 JSON Mode有的走 Function Calling但封装后对上层代码是透明的。Pydantic 模型定义越清楚字段描述越具体生成效果越好。3.2 工具定义与 Tool Calling工具是 Agent 连接外部世界的接口。LangChain 中定义一个工具非常简单from langchain_core.tools import tool tool def add(a: int, b: int) - int: 计算两个数字的和。 return a b工具函数的 docstring 非常重要因为模型会把它当作工具说明来理解“这个工具是干什么的、什么时候该用”。函数参数的类型和描述同样会被模型用来决定传什么值。实际项目中工具建议遵循以下设计原则单一职责一个工具只做一件事。参数约束尽量用基本类型 明确描述复杂对象用 JSON 字符串传入。失败反馈工具内部捕获异常并把清晰的错误信息返回给模型让模型有机会修正。3.3 状态管理与多轮执行循环Agent 不是一次调用就结束它需要循环模型生成回应 - 如果带工具调用就执行工具 - 把结果追加到消息历史 - 再次调用模型 - 直到没有工具调用为止。LangGraph 把这一过程建模为一个图Agent 节点接收当前状态调用模型。工具节点执行模型请求调用的工具。条件边判断模型是否有工具调用有则走工具节点无则结束。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] graph StateGraph(AgentState) def agent_node(state: AgentState): # 调用模型返回新的消息 pass def tool_node(state: AgentState): # 执行工具并把结果作为消息追加 pass graph.add_node(agent, agent_node) graph.add_node(tools, tool_node) graph.add_edge(tools, agent) graph.add_conditional_edges( agent, lambda state: tools if state[messages][-1].tool_calls else END )状态设计的核心是Annotated[list, add_messages]它保证每轮新增的消息会追加到消息列表而不是覆盖旧内容。这也是 Agent 能进行多轮工具调用而不丢失上下文的基础。3.4 记忆与上下文窗口控制Agent 的上下文窗口是有限的。TextToSQL 场景中如果每轮对话都把所有历史 SQL 和查询结果塞给模型很快就会把上下文挤爆。常用的策略有三种滑动窗口只保留最近 N 轮对话。摘要记忆把历史对话压缩成摘要再传回模型。持久化存储把关键信息存入外部数据库或向量库需要时检索。在 TextToSQL Agent 中我通常只保留最近几轮用户请求和 SQL 结果同时把数据库 schema 信息提前放入系统提示词因为它们比历史对话更重要。4. 实战用 LangChain LangGraph 实现 TextToSQL Agent接下来进入完整实战环节。我们的目标是构建一个能查询 SQLite 数据库的 Agent用户用自然语言提问Agent 自动生成 SQL、执行查询、返回结果。4.1 需求分析与项目结构先定义清楚功能边界只允许 SELECT 查询禁止 INSERT、UPDATE、DELETE、DROP。查询结果限制最多返回 100 条。如果 SQL 执行报错Agent 要能够根据错误信息自动修正。最终给用户的回复要包含查询结果和简要说明。text2sql-agent/ ├── .env ├── database.py # 数据库初始化与连接 ├── tools.py # 工具定义 ├── agent.py # Agent 构建与执行 └── main.py # 命令行入口4.2 初始化数据库先创建一个简单的销售数据库包含商品表、区域表和订单表-- 文件路径database.py 中执行 CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, category TEXT NOT NULL, price REAL NOT NULL ); CREATE TABLE IF NOT EXISTS regions ( id INTEGER PRIMARY KEY, region_name TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY, product_id INTEGER NOT NULL, region_id INTEGER NOT NULL, quantity INTEGER NOT NULL, order_date TEXT NOT NULL, FOREIGN KEY (product_id) REFERENCES products(id), FOREIGN KEY (region_id) REFERENCES regions(id) );写入部分示例数据# 文件路径database.py import sqlite3 DB_PATH sales.db def init_db(): conn sqlite3.connect(DB_PATH) cur conn.cursor() cur.executescript( CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, category TEXT NOT NULL, price REAL NOT NULL ); CREATE TABLE IF NOT EXISTS regions ( id INTEGER PRIMARY KEY, region_name TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY, product_id INTEGER NOT NULL, region_id INTEGER NOT NULL, quantity INTEGER NOT NULL, order_date TEXT NOT NULL ); ) # 插入初始数据如果表为空 if cur.execute(SELECT COUNT(*) FROM products).fetchone()[0] 0: cur.executemany(INSERT INTO products (name, category, price) VALUES (?, ?, ?), [(手机, 数码, 4999), (笔记本, 数码, 6999), (T恤, 服装, 99)]) cur.executemany(INSERT INTO regions (region_name) VALUES (?), [(华东,), (华北,), (华南,)]) cur.executemany(INSERT INTO orders (product_id, region_id, quantity, order_date) VALUES (?, ?, ?, ?), [(1, 1, 20, 2025-01-01), (2, 2, 10, 2025-01-02), (3, 3, 100, 2025-01-03)]) conn.commit() conn.close()这个初始化模块很简单但对后续演示很重要它保证了每个人跑出来的效果是一致的。4.3 定义数据库查询工具接下来定义 Agent 用到的工具。这里的关键是“只读安全”和“错误反馈”。# 文件路径tools.py import sqlite3 from langchain_core.tools import tool DB_PATH sales.db tool def get_db_schema(table_name: str ) - str: 获取数据库表结构信息。可以传入表名获取指定表结构不传则返回所有表。 conn sqlite3.connect(DB_PATH) cur conn.cursor() try: if table_name: tables [table_name] else: tables [row[0] for row in cur.execute(SELECT name FROM sqlite_master WHERE typetable).fetchall()] result [] for t in tables: cols cur.execute(fPRAGMA table_info({t})).fetchall() result.append(f表 {t}: , .join([col[1] for col in cols])) return \n.join(result) if result else 未找到相关表。 except Exception as e: return f获取表结构失败: {str(e)} finally: conn.close() tool def run_readonly_sql(sql: str) - str: 以只读模式执行一条 SQL 查询并返回结果。只能执行 SELECT 语句。 if not sql.strip().upper().startswith(SELECT): return 错误当前工具只支持 SELECT 查询。 conn sqlite3.connect(file:sales.db?modero, uriTrue) cur conn.cursor() try: # 去掉多余分号 sql_clean sql.strip().rstrip(;) # 强制限制最多返回100条 cur.execute(sql_clean LIMIT 100) columns [desc[0] for desc in cur.description] rows cur.fetchall() if not rows: return 查询执行成功但结果为空。 lines [, .join(columns)] for row in rows: lines.append(, .join(str(v) for v in row)) return \n.join(lines) except Exception as e: return fSQL 执行失败: {str(e)} finally: conn.close()这里有几个关键设计工具 docstring 写得很明确模型会根据这些说明决定何时调用工具。数据库连接使用modero以只读方式打开物理上杜绝写操作。自动追加LIMIT 100避免模型生成全表扫描的大查询。错误信息返回给模型而不是抛异常让 Agent 有机会根据错误修正 SQL。4.4 构建 Agent 执行链路来到最核心的部分用 LangGraph 构建 Agent 图。# 文件路径agent.py import os from typing import TypedDict, Annotated from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode from tools import get_db_schema, run_readonly_sql load_dotenv() class AgentState(TypedDict): messages: Annotated[list, add_messages] def build_agent(): llm ChatOpenAI( modelos.getenv(LLM_MODEL, gpt-4o-mini), api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL, https://api.openai.com/v1), temperature0, ) tools [get_db_schema, run_readonly_sql] llm_with_tools llm.bind_tools(tools) # 系统提示词说明 Agent 的角色与数据库结构 system_prompt 你是一个专业的数据库查询助手。你可以通过工具获取数据库表结构并编写 SQL 查询来回答用户问题。 请遵循以下规则 1. 先调用 get_db_schema 了解表结构再编写 SQL。 2. 只编写 SELECT 查询。 3. 条件值必须来自用户的问题不要凭空臆造。 4. 最终回答用简洁的中文说明查询结果不要输出多余内容。 def agent_node(state: AgentState): messages state[messages] # 如果第一条不是系统消息就插入系统提示词 if not messages or messages[0].type ! system: messages [{role: system, content: system_prompt}] messages response llm_with_tools.invoke(messages) return {messages: [response]} graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, ToolNode(tools)) graph.add_edge(tools, agent) def should_continue(state: AgentState): last_message state[messages][-1] return tools if getattr(last_message, tool_calls, None) else END graph.add_conditional_edges(agent, should_continue) graph.set_entry_point(agent) return graph.compile()这段代码的核心流程是用户输入先进入 agent 节点。模型根据工具描述决定调用get_db_schema还是run_readonly_sql。进入 tools 节点执行工具工具结果作为新消息追加回状态。回到 agent 节点模型看到工具结果后决定是继续调用工具还是输出最终回答。没有更多工具调用时流程结束。4.5 运行与验证最后写一个简单的命令行入口# 文件路径main.py from database import init_db from agent import build_agent def main(): init_db() agent build_agent() print(TextToSQL Agent 已启动输入问题后回车输入 exit 退出。) while True: user_input input(\n ) if user_input.strip().lower() exit: break result agent.invoke({messages: [{role: user, content: user_input}]}) answer result[messages][-1].content print(\n answer) if __name__ __main__: main()运行效果类似 华东区域的订单里销量最高的商品是什么Agent 会先调用get_db_schema了解结构再生成类似下面的 SQLSELECT p.name, SUM(o.quantity) AS total_quantity FROM orders o JOIN products p ON o.product_id p.id JOIN regions r ON o.region_id r.id WHERE r.region_name 华东 GROUP BY p.name ORDER BY total_quantity DESC LIMIT 1因为工具限制了只读、错误自动反馈、结果自动加 LIMIT整个流程就算生成出不合法的 SQLAgent 也能看到错误信息并重新尝试。5. 常见问题与排查思路5.1 高频报错与解决方案问题现象常见原因解决思路模型一直调用工具不结束工具结果不够明确模型不知道任务已完成让工具返回结构化清晰的结果并在系统提示词中明确“查询完直接回答”Agent 生成的 SQL 语法总是错误模型不了解数据库方言把 SQL 方言、示例查询放入系统提示词或提供少量 Few-Shot 示例查询结果太大上下文溢出SQL 没有 LIMIT工具层强制追加 LIMIT或对结果做截断处理模型拒绝调用任何工具工具注册失败或模型版本不支持 Tool Calling检查bind_tools是否生效换用支持 Function Calling 的模型输出格式不稳定没有使用结构化输出使用with_structured_output或要求 JSON 输出5.2 排查 Checklist如果你在跑 Agent 时遇到问题建议按下面的顺序排查先确认模型本身能正常调用不经过 Agent只做一次llm.invoke测试。确认工具函数能否单独执行避免 Agent 报错是工具内部 Bug。打印整个消息历史看模型每轮收到的上下文是什么。检查工具返回的文本长度太长可能影响模型判断。看 LangGraph 图的每个节点执行路径确认是停留在 agent 还是 tools。这里特别提醒一个新手容易踩的坑很多 Agent 问题不是模型不行而是工具返回格式太杂乱。工具返回值就是喂给模型的“事实”如果事实不清晰模型再聪明也会失误。6. 最佳实践与工程建议6.1 安全边界TextToSQL 是高风险场景TextToSQL 最怕的不是 SQL 写错而是 SQL 做了不该做的事。即使模型不会故意攻击用户输入也可能携带恶意指令导致生成危险 SQL。在生产环境至少要做到这几点数据库账号最小化权限使用只读账号甚至单独建一个只能查视图的账号。连接层只读保护像示例中那样使用modero避免代码失误造成写操作。SQL 白名单校验在工具层做关键词拦截只放行 SELECT。查询限流与超时防止某一个查询占用过多资源。敏感字段脱敏如果表里有手机号、身份证等字段查询前需要做列级权限控制。6.2 可观测性与评估Agent 的决策链路比较长模型生成 SQL 到执行之间任何一个环节出错都可能导致业务同学拿不到数据。所以日志追踪非常重要。建议在每个环节埋点记录环节记录内容用户输入原始自然语言问题模型输出是否调用工具、调用哪个工具、传入参数工具执行SQL 语句、执行耗时、行数最终回答返回给用户的文本本地开发可以先打日志生产环境建议接入 OpenTelemetry 或 LangSmith 这类可观测平台。有了日志之后再去做效果评估才有依据。TextToSQL 的评估集也很重要。准备 50~100 条典型问题标注正确答案然后定期跑回归看多少比例的 SQL 结果是正确的。没有评估集优化就全凭感觉。6.3 成本控制与性能优化Agent 的调用次数比普通聊天应用多得多。一次查询可能触发两三次甚至更多次模型调用。控制成本可以从几方面入手用小模型做简单分类先判断用户的问题是否涉及数据库查询不涉及就走普通对话链路。复用 schema 信息把表结构预先缓存不必每轮都调用工具去取。控制历史消息数量只保留最近 1~2 轮减少输入 token。使用模型路由简单问题用便宜的小模型复杂 SQL 才调用强模型。6.4 Harness 落地的三步建议前面提到 Harness 工程这里给出具体的落地建议。不要一开始就追求大而全的 Agent 平台从最小的三层做起第一层是模型层。统一模型 API 代理屏蔽底层供应商切换的影响。第二层是工具层。把所有工具做注册、鉴权、限流、审计。每个工具都要有负责人、超时时间和失败兜底。第三层是运行时层。封装 LangGraph 的输入输出格式统一处理用户会话、日志追踪和错误码。这样业务方对接时不需要关心底层调用了哪个模型、用了什么框架。这三层合起来就是一个小型的 Agent Harness。先跑通再扩展比直接引一套重型平台更容易落地。7. 总结与下一步学习路线这篇文章从大模型 Agent 的概念出发介绍了 LangChain 和 LangGraph 的分工解释了 Harness 工程在 LLM 应用交付中的位置然后用一个完整的 TextToSQL 项目把工具定义、状态管理、执行循环串了起来。你至少应该掌握三件事第一Agent 的核心不是模型本身而是“模型 工具 状态循环”的组合第二LangChain 封装基础组件LangGraph 编排复杂流程第三TextToSQL 这类业务场景真正的难点在安全管控、结果可评估、失败可恢复而不是 SQL 生成本身。下一步的建议是先把这个 SQLite 示例跑通然后把它改造成连接真实业务库注意权限和备份再尝试加入多轮追问能力比如用户说“换成华东再看看”Agent 能结合上一轮上下文重新查询。更进一步可以学 RAG、记忆机制、多 Agent 协作。如果这篇文章对你有帮助可以收藏备用。也欢迎在评论区聊聊你在 Agent 开发中踩过的坑特别是 TextToSQL 相关的安全和效果问题互相交流。