LangChain与LangGraph生态下Agent全栈开发：Harness工程与TextToSQL实践

实际项目里大模型 Agent 开发早已不是“调一次 API、拼一个 Prompt”这么简单。真正的分水岭出现在任务需要多步决策的时候Agent 要决定调哪个工具、工具调用失败了怎么办、循环次数怎么控制、日志能不能追踪到每一次模型决策。这篇文章围绕 LangChain V1.0 生态、Harness 工程、智能体工具和 TextToSQL 项目落地这条主线带读者从概念到代码完整走一遍 Agent 全栈开发。适合具备 Python 基础、已经调过大模型 API、但还在 Demo 阶段徘徊的开发者。学完后可以把这个骨架迁移到报表问答、企业知识库和运维诊断等真实业务场景。文章会从最小可运行代码开始逐步加入 Harness 工程能力最后给出生产环境必须关注的超时、权限、日志和排查路径。所有代码都是为了说明思路实际项目里要结合自己的模型服务、数据库方言和包版本做调整。1. 先搞清楚Agent 开发和大模型调用不是一回事1.1 单次问答与多步任务的本质差异一次普通的大模型调用是“输入文本 - 输出文本”的单向过程。模型拿到用户问题根据参数直接生成回答整个过程只有一步结果不可执行、不可验证。Agent 处理的问题是另一个类型用户问“查询上个月销量前五的商品名称和总金额”模型无法单靠参数知识回答它需要先看数据库里有哪些表再决定写什么 SQL执行 SQL拿到结果后还要整理成自然语言回答。这个流程里存在多次模型决策和多次外部调用中间任何一步失败都要能反馈给模型重新尝试。这就是 Agent 与普通模型调用的核心差异普通调用用户输入 - 模型输出 Agent用户输入 - 模型决策 - 工具调用 - 观察结果 - 再次决策 - 最终输出理解了这条链路就能明白为什么 Agent 开发不能只关注 Prompt 写得漂不漂亮还要关注工具边界、循环控制、超时、日志和权限。这些内容在单次模型调用里都不存在但在 Agent 项目里会成为主要故障来源。1.2 Agent 系统的四个核心部件一个可用的 Agent 系统通常由四部分组成缺一不可部件职责常见实现模型负责推理、理解和决策ChatOpenAI、本地 Ollama 或 vLLM 部署的模型工具把外部能力暴露给模型调用tool 定义的函数、HTTP API、数据库查询编排层控制“决策-执行-观察”循环LangGraph、create_react_agentHarness工程护栏超时、重试、权限、日志、限额自研包装层或 SDK 提供的运行时外壳在许多人写的 Demo 里只有前三部分Harness 是完全缺失的。模型一旦陷入死循环或者工具调用迟迟不返回程序就卡死在那里没有任何观测手段。Harness 解决的正是这一类问题。1.3 Harness 工程Agent 的“运行外壳”Harness 这个词在 Agent 开发里通常指 Agent 的运行时外壳也就是承载 Agent 生命周期、对外交互、工具调度、安全约束和观测能力的工程层。可以这样理解模型是 Agent 的“脑子”编排层是“神经系统”Harness 是“安全带、仪表盘和刹车”。为什么要单独强调 Harness因为大模型天然存在三个工程问题模型可能臆造工具参数或 SQL 字段名需要工具层做严格校验。工具调用可能长时间不返回需要超时和熔断。Agent 循环可能失控需要限制递归次数、记录每一次决策。这些都不能靠 Prompt 解决必须靠工程代码兜底。这也是“Harness 和 Agent 的区别”这个高频问题背后的答案Agent 解决“能不能做”Harness 解决“做的时候会不会出事、出事了能不能查”。2. LangChain V1.0 生态LangChain 与 LangGraph 的分工2.1 先理解 V1.0 时代的包组织方式LangChain V1.0 发布后生态的组织方式比早期版本清晰很多。核心思路是分层LangChain 负责提供模型、提示词、工具等组件LangGraph 负责编排和控制循环模型供应商的适配能力拆分到独立包中例如langchain-openai。实际写代码时最常见的导入路径是from langchain_openai import ChatOpenAI # 模型适配层 from langchain_core.prompts import ChatPromptTemplate # 提示词组件 from langchain_core.tools import tool # 工具定义 from langgraph.prebuilt import create_react_agent # 编排层同一个项目里如果同时用到langchain、langgraph、langchain-openai要保证版本在同一代际否则很容易出现导入路径不对、参数签名不一致的问题。V1.0 之后网上的旧教程大量失效其中一个原因就是旧代码使用了已经迁移的导入路径。这里有一个非常实用的判断方法落地前先查看当前安装的真实版本再按该版本的官方文档调整代码不要盲目复制旧博客里的调用方式。2.2 LangChain 与 LangGraph 的区别与选型“LangChain 和 LangGraph 的区别”是被问得最多的问题之一。可以这样区分对比项LangChainLangGraph定位组件库编排框架核心能力模型、Prompt、Tool、RAG 组件状态图、节点、边、持久化解决什么问题把常用能力封装成可复用组件控制 Agent 的多步循环和状态流转典型 APIChatOpenAI、ChatPromptTemplate、toolStateGraph、create_react_agent适用场景轻量链式调用、RAG 管道需要工具调度、多轮记忆、分支控制的 Agent选型建议很简单如果只是静态的“取问题-检索-生成回答”链路用 LangChain 组件就够了。一旦出现“让模型决定要不要调工具、调完工具再决定下一步”这种循环逻辑就应该进入 LangGraph而不是自己在 LangChain 里写 while 循环。2.3 环境准备与项目骨架先创建虚拟环境并安装依赖mkdir agent_text2sql cd agent_text2sql python -m venv .venv source .venv/bin/activate pip install langchain1.0 langgraph langchain-openai1.0 sqlalchemy pydantic python-dotenvWindows 下激活命令是.venv\Scripts\activate。安装完成后按下面结构组织代码agent_text2sql/ ├── .env.example ├── requirements.txt └── app/ ├── __init__.py ├── database.py ├── schema_provider.py ├── tools.py ├── harness.py └── agent.py.env.example内容如下LLM_API_KEYsk-xxx LLM_BASE_URL LLM_MODELgpt-4o-mini DATABASE_URLsqlite:///./sales.db注意模型接口和数据库地址不要写死在代码里。学习阶段可以用.env生产环境必须接入密钥管理系统。如果当前环境无法访问外部模型服务可以把LLM_BASE_URL指向本地部署的模型网关例如 Ollama 或 vLLM 提供的兼容 OpenAI 协议的服务。开发阶段最重要的是把链路跑通模型本身可以后续再换。3. 从 LangChain 到 LangGraph先跑通最小 Agent 骨架3.1 初始化模型服务用ChatOpenAI初始化模型这是 LangChain 生态里最常见的模型接入方式import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( modelos.getenv(LLM_MODEL, gpt-4o-mini), temperature0, api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL) or None, timeout60, )这里有两个关键参数。temperature0用于 SQL 生成场景让模型输出更确定不要有随机发挥。timeout60是会话超时防止模型服务端不返回时请求无限挂起。在 Agent 项目里模型超时是第一个要设置的 Harness 能力因为一次 Agent 运行可能包含多次模型调用任何一次卡住都会拖垮整个任务。3.2 第一个 ReAct AgentReAct 是 Agent 最常见的工作模式Reason 和 Act 交替进行模型先思考要做什么再调用工具观察结果后继续思考。用 LangGraph 的create_react_agent可以直接得到一个完整的 ReAct Agentfrom langchain_core.tools import tool from langgraph.prebuilt import create_react_agent tool def get_current_time() - str: 返回当前系统时间用于回答与时间相关的问题。 import datetime return datetime.datetime.now().isoformat() agent create_react_agent(llm, tools[get_current_time]) result agent.invoke({ messages: [{role: user, content: 现在几点了}] }) print(result[messages][-1].content)create_react_agent内部已经封装了“决策-调用-观察-再决策”的循环。传给它的tools列表会被自动转换成模型可理解的工具描述模型在需要的时候会返回tool_calls框架负责执行工具并把结果作为消息放回会话里。3.3 让 Agent 带上对话记忆Agent 的多轮对话不是简单把历史消息拼回去更好的做法是使用检查点机制。LangGraph 里的MemorySaver可以在内存中保存会话状态from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() agent create_react_agent( llm, tools[get_current_time], checkpointermemory, ) result agent.invoke( {messages: [{role: user, content: 我 1995 年出生今年多大了}]}, config{configurable: {thread_id: thread-1}}, )thread_id用来区分不同会话。同一个thread_id的多次调用共享记忆不同用户或不同任务必须使用不同thread_id。生产环境不建议用MemorySaver因为它只存在内存里进程重启就丢失。需要持久化记忆时应切换为基于数据库的检查点实现例如 PostgreSQL 版 Checkpointer。3.4 检查点如何确认 Agent 真的调用了工具只看最终输出无法判断 Agent 是否真的调用了工具。调试时把整条消息列表打出来for msg in result[messages]: tool_calls getattr(msg, tool_calls, None) print(msg.type, tool_calls)预期会看到类似这样的序列AIMessage [{name: get_current_time, args: {}, ...}] ToolMessage None AIMessage None如果输入一个明显需要时间的问题但整个过程中没有任何tool_calls问题通常不是模型能力不够而是工具描述写得不好或者参数 schema 定义得太模糊。这一点在下一节工具设计里会重点展开。4. Harness 工程给 Agent 装上一层可控和可观测的运行外壳4.1 为什么不能只靠递归限制create_react_agent默认有递归次数限制但光靠框架内置参数还不够。模型请求会超时工具会挂起模型可能连续调用同一个错误工具多次这些都需要在 Harness 层统一处理。先看学习环境和生产环境的能力差异控制项学习环境生产环境模型超时60 秒分级超时普通请求 20 秒复杂任务 90 秒工具调用失败直接抛异常指数退避重试最多 2 次循环次数框架默认 recursion_limit按业务场景定义明确上限日志printJSON 结构化日志 trace 指标告警权限全量放开最小权限、只读账号、行级权限成本控制无单任务 token 上限、每月调用配额4.2 实现一个轻量 Harness 包装器不引入额外框架也可以先写一个极简包装器把超时、配置和日志收敛到同一处import json import logging import time logger logging.getLogger(agent_harness) class AgentHarness: def __init__(self, agent, *, max_seconds30, recursion_limit25): self.agent agent self.max_seconds max_seconds self.recursion_limit recursion_limit def run(self, user_text: str, thread_id: str default): start time.time() inputs { messages: [{role: user, content: user_text}], } config { recursion_limit: self.recursion_limit, configurable: {thread_id: thread_id}, } result self.agent.invoke(inputs, configconfig) elapsed time.time() - start if elapsed self.max_seconds: raise TimeoutError( fagent run timeout: {elapsed:.2f}s {self.max_seconds}s ) self._log(result, elapsed) return result def _log(self, result, elapsed): trace [] for msg in result[messages]: if getattr(msg, tool_calls, None): call msg.tool_calls[0] trace.append({ type: tool_call, name: call[name], args: call[args], }) elif msg.type tool: trace.append({ type: tool_result, name: msg.name, content: str(msg.content)[:200], }) logger.info(json.dumps({ latency_ms: round(elapsed * 1000), message_count: len(result[messages]), trace: trace, }, ensure_asciiFalse))这个包装器的价值在于所有 Agent 调用都从同一个入口进出超时判断、日志格式、递归限制都集中管理。后续要加熔断、限流、成本统计也只需要在这个类里扩展。4.3 结构化日志和 trace 怎么排查问题上面的_log