实际项目里，大模型 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_agent |
| Harness | 工程护栏：超时、重试、权限、日志、限额 | 自研包装层或 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 的区别”是被问得最多的问题之一。可以这样区分：
| 对比项 | LangChain | LangGraph |
|---|---|---|
| 定位 | 组件库 | 编排框架 |
| 核心能力 | 模型、Prompt、Tool、RAG 组件 | 状态图、节点、边、持久化 |
| 解决什么问题 | 把常用能力封装成可复用组件 | 控制 Agent 的多步循环和状态流转 |
| 典型 API | ChatOpenAI、ChatPromptTemplate、@tool | StateGraph、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 "langchain>=1.0" langgraph "langchain-openai>=1.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_KEY=sk-xxx LLM_BASE_URL= LLM_MODEL=gpt-4o-mini DATABASE_URL=sqlite:///./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( model=os.getenv("LLM_MODEL", "gpt-4o-mini"), temperature=0, api_key=os.getenv("LLM_API_KEY"), base_url=os.getenv("LLM_BASE_URL") or None, timeout=60, )这里有两个关键参数。temperature=0用于 SQL 生成场景，让模型输出更确定，不要有随机发挥。timeout=60是会话超时，防止模型服务端不返回时请求无限挂起。在 Agent 项目里，模型超时是第一个要设置的 Harness 能力，因为一次 Agent 运行可能包含多次模型调用，任何一次卡住都会拖垮整个任务。
3.2 第一个 ReAct Agent
ReAct 是 Agent 最常见的工作模式：Reason 和 Act 交替进行，模型先思考要做什么，再调用工具，观察结果后继续思考。
用 LangGraph 的create_react_agent可以直接得到一个完整的 ReAct Agent：
from 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], checkpointer=memory, ) 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 | 按业务场景定义明确上限 |
| 日志 | JSON 结构化日志 + trace + 指标告警 | |
| 权限 | 全量放开 | 最小权限、只读账号、行级权限 |
| 成本控制 | 无 | 单任务 token 上限、每月调用配额 |
4.2 实现一个轻量 Harness 包装器
不引入额外框架，也可以先写一个极简包装器，把超时、配置和日志收敛到同一处：
import json import logging import time logger = logging.getLogger("agent_harness") class AgentHarness: def __init__(self, agent, *, max_seconds=30, recursion_limit=25): 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, config=config) elapsed = time.time() - start if elapsed > self.max_seconds: raise TimeoutError( f"agent 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_ascii=False))这个包装器的价值在于，所有 Agent 调用都从同一个入口进出，超时判断、日志格式、递归限制都集中管理。后续要加熔断、限流、成本统计，也只需要在这个类里扩展。
4.3 结构化日志和 trace 怎么排查问题
上面的_log
