LangChain实战：从Toolkit到Python Agent与SQL Agent的演进指南-尧图网络科技

1. 项目概述：从工具函数到智能体的演进之路

最近在折腾LangChain，发现很多朋友对它的Toolkit和Agent概念有点模糊，尤其是从简单的工具函数调用，到能自主决策的Python Agent，再到能直接操作数据库的SQL Agent，这中间的路径和实战细节，网上资料要么太散，要么直接丢给你一个看不懂的复杂例子。今天我就结合自己踩过的坑，把这套“工具链”的实战进化史捋清楚。你完全可以把它看作一个能力逐步增强的“AI员工”培养手册：最开始它只会你教的一个固定动作（工具函数），后来学会了根据你的指令从多个动作里选一个（Python Agent），最后甚至能直接去数据库里翻箱倒柜帮你找答案（SQL Agent）。无论你是想快速给现有应用加个智能对话功能，还是想构建一个能自动分析数据的AI助手，这套从简到繁的思路都能给你一个清晰的起点。

2. 核心基石：深入理解LangChain Toolkit

在谈论Agent之前，我们必须先夯实地基，也就是Toolkit。你可以把它理解为一个“AI可用的工具箱”。但别误会，这不是简单地把你的Python函数打个包就行。一个设计良好的Tool，是Agent能够可靠工作的前提。

2.1 Tool的本质与设计规范

LangChain中的Tool，本质上是一个将自然语言描述与可执行代码（函数）进行绑定的标准化接口。大模型（LLM）通过这个描述来理解“这个工具能干什么”，并在需要时调用背后的函数。创建一个Tool，远不止是写个@tool装饰器那么简单。

首先，描述（description）是灵魂。一个模糊的描述会导致LLM误用或根本想不到用它。比如，一个查询天气的函数，糟糕的描述是：“获取天气信息”。而好的描述应该是：“根据提供的城市名称，查询该城市当前的天气状况，包括温度、天气现象（晴、雨等）、湿度和风速。输入应为单个字符串格式的城市名。”

其次，参数处理要健壮。LLM传来的参数可能是字符串、字典，甚至是不完整的JSON。你的函数内部必须做好类型校验、默认值处理和异常捕获。我习惯在工具函数内部一开始就进行参数解析和清洗，确保核心逻辑拿到的是干净、结构化的数据。

from langchain.tools import tool from typing import Optional import requests @tool def get_weather(city_name: str) -> str: """ 根据城市名称查询实时天气。 参数: city_name (str): 城市的名称，例如“北京”、“Shanghai”。 返回: str: 格式化的天气信息字符串，包含温度、天气、湿度等。 """ # 1. 参数清洗与验证 if not city_name or not isinstance(city_name, str): return “请输入有效的城市名称。” city_name = city_name.strip() # 2. 核心业务逻辑（这里用模拟数据代替真实API调用） # 在实际项目中，这里会调用如OpenWeatherMap的API weather_data = { “temperature”: “22°C”, “conditions”: “晴”, “humidity”: “65%”, “wind_speed”: “10 km/h” } # 3. 格式化返回，便于LLM理解和后续展示 return f“{city_name}的天气情况：温度{weather_data[‘temperature’]}，{weather_data[‘conditions’]}，湿度{weather_data[‘humidity’]}，风速{weather_data[‘wind_speed’]}。” # 测试工具 print(get_weather.invoke({“city_name”: “北京”}))

注意：Tool函数的返回值最好是结构清晰的字符串。虽然LLM能解析复杂JSON，但清晰的文本更利于它生成流畅的自然语言回复。避免返回原生Python对象（如字典、列表）而不做任何处理。

2.2 构建你的第一个工具箱（Toolkit）

单个工具能力有限，通常我们需要把相关工具组合成一个Toolkit，供Agent选择。例如，一个“数据查询工具箱”可能包含：查询天气、查询股票价格、查询汇率。在LangChain中，Toolkit就是一组Tool的集合。

创建Toolkit的关键在于功能的内聚性。把毫不相干的工具塞进一个工具箱，只会让Agent感到困惑。好的做法是按领域划分：数据分析工具箱、文件操作工具箱、网络搜索工具箱等。

from langchain.agents import create_toolkit # 假设我们已经定义了多个工具 weather_tool = get_weather # 上面的天气工具 stock_tool = get_stock_price # 假设的股票查询工具 currency_tool = get_exchange_rate # 假设的汇率查询工具 # 将这些工具组合成一个工具箱 data_query_toolkit = [weather_tool, stock_tool, currency_tool] # 在实际创建Agent时，我们会直接传递这个工具列表

这里有一个高级技巧：为工具设计优先级或依赖关系。虽然LangChain的Agent会自己决定使用哪个工具，但在某些场景下，你可以通过提示词（Prompt）来隐式引导。例如，在提示词中强调“当用户询问金钱相关问题时，优先考虑使用汇率查询工具”。

3. Python Agent实战：让AI学会“思考”与“选择”

有了工具箱，我们就可以进入下一个阶段：创建Python Agent。Agent与单纯工具调用的最大区别在于引入了“思考链”（ReAct模式：Reasoning + Acting）。Agent会根据你的问题，自主决定是否需要使用工具、使用哪个工具、以及如何解读工具的返回结果。

3.1 Agent的核心工作流与ReAct模式

当你向一个配备了工具的Python Agent提问时，它内部的工作流是这样的：

理解问题：LLM解析你的输入。
制定计划：LLM判断是否需要使用工具来解决问题。如果需要，它会“思考”应该选用哪个工具，并生成调用该工具所需的参数。
执行行动：Agent框架调用被选中的工具，并传入参数。
观察结果：工具执行完毕，返回结果给Agent。
反思与迭代：LLM根据工具返回的结果，判断问题是否已解决。如果未解决，则重复步骤2-4，可能选择其他工具或调整参数；如果已解决，则综合所有信息生成最终答案。

这个“思考-行动-观察”的循环，就是ReAct模式的核心。它让AI不再是一次性输出，而是具备了多步推理和交互能力。

3.2 使用`create_agent`函数构建智能体

create_agent是构建Agent的一种高级、简洁的方式。它帮你封装了Agent执行器（AgentExecutor）的创建过程，让你更关注工具和LLM本身。

from langchain import hub from langchain.agents import create_agent, AgentExecutor from langchain.agents.format_scratchpad import format_log_to_str from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain.tools.render import render_text_description from langchain_openai import ChatOpenAI import os # 0. 设置你的LLM，这里以OpenAI为例 os.environ[“OPENAI_API_KEY”] = “your-api-key-here” llm = ChatOpenAI(model=“gpt-3.5-turbo”, temperature=0) # 1. 准备工具列表（使用之前定义的data_query_toolkit） tools = data_query_toolkit # 2. 获取一个预设的ReAct风格提示词模板 # LangChain Hub上有很多社区贡献的优质提示词 prompt = hub.pull(“hwchase17/react”) # 3. 绑定工具描述到提示词中 # 这一步至关重要，让LLM知道它有哪些工具可用 prompt = prompt.partial( tools=render_text_description(tools), # 将工具列表渲染成文本描述 tool_names=“, “.join([t.name for t in tools]) # 提供工具名称列表 ) # 4. 定义Agent的运行逻辑 llm_with_stop = llm.bind(stop=[“\nObservation:”]) # 告诉LLM在哪里停止生成，以等待工具执行结果 # 5. 构建Agent的推理链路 agent = ( { “input”: lambda x: x[“input”], “agent_scratchpad”: lambda x: format_log_to_str(x[“intermediate_steps”]), # 格式化执行历史 } | prompt # 输入经过提示词模板 | llm_with_stop # 送入LLM生成思考过程 | ReActSingleInputOutputParser() # 解析LLM输出，提取工具调用指令或最终答案 ) # 6. 创建Agent执行器 agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True) # 7. 运行Agent result = agent_executor.invoke({“input”: “请问北京现在的天气怎么样，同时100美元能换多少人民币？”}) print(result[“output”])

当你运行这段代码并将verbose设为True时，你会在控制台看到完整的思考过程：

> Entering new AgentExecutor chain... 我需要回答两个问题：北京的天气和美元兑人民币的汇率。我可以先用天气工具查询北京天气，再用汇率工具查询美元兑人民币汇率。 Action: get_weather Action Input: {“city_name”: “北京”} Observation: 北京的天气情况：温度22°C，晴，湿度65%，风速10 km/h。 我现在有了天气信息。接下来需要汇率信息。 Action: get_exchange_rate Action Input: {“from_currency”: “USD”, “to_currency”: “CNY”, “amount”: 100} Observation: 100美元可兑换约718.5人民币。 现在我综合这两个信息来回答用户。 > Finished chain. 北京目前天气晴朗，温度22摄氏度，湿度65%，风速10公里/小时。另外，根据当前汇率，100美元大约可以兑换718.5元人民币。

实操心得：handle_parsing_errors=True这个参数非常有用。LLM有时生成的工具调用格式可能不规范，设置这个参数能让执行器尝试自动修复，而不是直接崩溃。这在生产环境中能显著提高系统的鲁棒性。

3.3 提示词工程：引导Agent更精准地工作

Agent的表现严重依赖于提示词。上面例子中从Hub拉取的react提示词是个不错的起点，但对于复杂任务，你通常需要自定义。核心是在提示词中明确以下几点：

角色定义：告诉AI它扮演什么角色（例如，“你是一个专业的数据分析助手”）。
工具说明书：清晰列出每个工具的名称、描述、输入格式和输出示例。
约束与规则：规定它必须使用工具、不能编造信息、如何格式化输出等。
思考格式：明确要求它按照“Thought:”, “Action:”, “Action Input:”, “Observation:”的格式进行推理。

一个自定义的提示词模板可能长这样：

你是一个智能助手，可以调用以下工具来帮助用户： {tools} 请严格按照以下格式回应： Thought: 你需要思考现在应该做什么 Action: 需要调用的工具名称，必须是[{tool_names}]中的一个 Action Input: 调用该工具所需的输入，必须是有效的JSON格式 Observation: 工具返回的结果 当你得出最终答案时，请以“Final Answer:”开头。 开始！ 用户问题：{input} {agent_scratchpad}

4. SQL Agent实战：让AI直接与数据库对话

如果说Python Agent是让AI调用通用API，那么SQL Agent就是专门为数据库操作而生的“专家”。它允许你用自然语言查询数据库，AI会自动生成SQL语句、执行、并解释结果。这对于不会SQL的业务人员，或者需要快速进行数据探查的开发者来说，是革命性的工具。

4.1 为何需要专门的SQL Agent？

你可能会问，用普通的Python Agent，加一个“执行SQL”的工具不就行了吗？理论上可以，但实践中有诸多挑战：

数据库Schema复杂：LLM需要理解表结构、字段类型、关联关系。
SQL语法与安全：生成的SQL必须语法正确，且要防止SQL注入等安全问题。
结果解释：直接返回数据库查询结果（如元组列表）对用户不友好，需要转换成自然语言。

LangChain的SQL Agent通过create_sql_agent函数，内置了一套专门处理这些问题的机制。它集成了SQLDatabase Toolkit，这个工具箱里包含了描述表结构、查询示例、执行查询、检查查询结果等多个协同工作的工具。

4.2 构建你的第一个SQL Agent

让我们一步步构建一个连接SQLite数据库的Agent。

from langchain_community.utilities import SQLDatabase from langchain_community.agent_toolkits import create_sql_agent from langchain_openai import ChatOpenAI import sqlite3 # 1. 创建或连接一个示例数据库 conn = sqlite3.connect(‘example.db’) cursor = conn.cursor() # 创建一个简单的员工表 cursor.execute(‘’’CREATE TABLE IF NOT EXISTS employees (id INTEGER PRIMARY KEY, name TEXT, department TEXT, salary REAL)’’’) # 插入一些示例数据 cursor.executemany(‘INSERT INTO employees (name, department, salary) VALUES (?, ?, ?)’, [(‘张三’, ‘技术部’, 15000), (‘李四’, ‘销售部’, 12000), (‘王五’, ‘技术部’, 18000), (‘赵六’, ‘人事部’, 9000)]) conn.commit() # 2. 创建SQLDatabase对象，这是LangChain与数据库交互的抽象层 db = SQLDatabase.from_uri(“sqlite:///example.db”) # 3. 初始化LLM llm = ChatOpenAI(model=“gpt-3.5-turbo”, temperature=0) # 4. 创建SQL Agent agent_executor = create_sql_agent( llm=llm, db=db, agent_type=“openai-tools”, # 使用OpenAI函数调用格式的Agent，更稳定 verbose=True, handle_parsing_errors=True ) # 5. 用自然语言提问！ result = agent_executor.invoke( {“input”: “技术部有多少名员工？他们的平均工资是多少？”} ) print(result[“output”])

运行后，你会看到Agent的思考过程：它先调用工具查看数据库中有哪些表（sql_db_list_tables），然后查看表结构（sql_db_schema），接着生成并执行SQL（sql_db_query），最后对结果进行总结。

4.3 高级技巧与安全考量

直接让AI生成并执行SQL听起来很强大，但也非常危险。以下是我在实际项目中总结的必须遵守的准则：

严格的数据库连接权限：永远不要给Agent一个具有DROP、DELETE、UPDATE权限的数据库账户。应该创建一个只读（SELECT）权限的专用账户。在SQLite中，这意味着以只读模式打开连接，或在生产数据库中严格限制权限。
使用自定义提示词限制查询范围：在create_sql_agent中，你可以传入自定义的prompt参数。务必在提示词中强调：“你只能执行SELECT查询，严禁执行任何数据修改（INSERT, UPDATE, DELETE, DROP等）操作。” 虽然LLM大多会遵守，但这是一个重要的安全层。
结果行数限制：避免Agent执行一个返回百万行数据的查询拖垮数据库。可以在SQLDatabase初始化时设置sample_rows_in_table_info参数，限制它查看的样本行数，也可以在提示词中要求“如果结果超过100行，请只进行汇总分析”。
处理复杂查询与错误：对于多表JOIN或复杂子查询，LLM可能会生成错误SQL。create_sql_agent的好处在于，当执行出错时，Agent会将错误信息作为Observation反馈给LLM，LLM有机会修正SQL后重试。将handle_parsing_errors和max_iterations（最大重试次数）参数搭配使用，可以提高成功率。

agent_executor = create_sql_agent( llm=llm, db=db, agent_type=“openai-tools”, verbose=True, handle_parsing_errors=True, max_iterations=5, # 限制最大重试次数，避免死循环 early_stopping_method=“generate”, # 设置提前停止策略 agent_executor_kwargs={“handle_parsing_errors”: True} # 双重保险 )

5. 常见问题排查与性能优化实录

在实际开发和部署Agent的过程中，你一定会遇到各种问题。下面是我整理的一些典型“坑”及其解决方案。

5.1 Agent陷入循环或拒绝使用工具

现象：Agent一直在“思考”，但就是不调用工具；或者反复调用同一个工具，无法得出最终答案。根因：

工具描述不清晰：LLM无法准确理解工具用途。
提示词约束过强或过弱：可能没有强制要求它使用工具，或者没有给出停止思考的明确指令。
LLM温度（temperature）设置过高：导致输出随机性太大，无法稳定遵循指令。解决方案：

仔细打磨工具描述，确保无歧义，并包含输入输出示例。
在提示词中明确写出：“你必须使用提供的工具来回答问题。如果你认为工具无法解决，请直接说‘我无法用现有工具回答这个问题’。”
将LLM的temperature参数调低（如设为0），以获得更确定性的输出。

5.2 SQL Agent生成错误或危险的SQL语句

现象：生成的SQL语法错误，或者试图执行DELETE语句。根因：

Agent对数据库Schema理解不准确。
提示词中安全约束不足。解决方案：

确保SQLDatabase对象能正确获取表结构信息。对于大型数据库，可以使用custom_table_info参数手动提供关键表的精简Schema，避免信息过载。
实施强制安全策略：这是最重要的。不要依赖LLM的自觉性。在应用层，对Agent生成的SQL语句进行静态检查。可以使用简单的正则表达式或SQL解析库（如sqlparse）在执行前过滤掉所有非SELECT的关键字。

import re import sqlparse def is_select_query(sql: str) -> bool: “”“检查SQL是否为安全的SELECT查询”“” parsed = sqlparse.parse(sql) if not parsed: return False first_token = parsed[0].token_first(skip_cm=True) return first_token and first_token.value.upper() == ‘SELECT’ # 在执行SQL前进行拦截 if not is_select_query(generated_sql): raise ValueError(“只允许执行SELECT查询！”)

5.3 处理复杂、多轮对话的上下文

现象：在连续对话中，Agent忘记了之前的对话历史，导致每次回答都像重新开始。根因：默认的Agent执行器是“无状态”的，每次invoke都是独立的。解决方案：你需要引入记忆（Memory）组件。LangChain提供了多种记忆后端，如ConversationBufferMemory。关键是将记忆整合到Agent的输入输出循环中。

from langchain.memory import ConversationBufferMemory from langchain.agents import AgentExecutor # 创建记忆体 memory = ConversationBufferMemory(memory_key=“chat_history”, return_messages=True) # 在创建提示词时，加入记忆变量 prompt = hub.pull(“hwchase17/react-chat”) prompt = prompt.partial( tools=render_text_description(tools), tool_names=“, “.join([t.name for t in tools]), chat_history=“{chat_history}” # 在提示词模板中预留位置 ) # 重新定义Agent的输入，包含记忆 agent = ( { “input”: lambda x: x[“input”], “chat_history”: lambda x: x.get(“chat_history”, “”), # 从输入中获取历史 “agent_scratchpad”: lambda x: format_log_to_str(x[“intermediate_steps”]), } | prompt | llm_with_stop | ReActSingleInputOutputParser() ) # 创建执行器时传入记忆 agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, memory=memory, # 关键：绑定记忆 handle_parsing_errors=True ) # 现在可以进行多轮对话了 result1 = agent_executor.invoke({“input”: “北京天气如何？”}) print(result1[“output”]) result2 = agent_executor.invoke({“input”: “那上海呢？”}) # Agent会记得之前聊过天气 print(result2[“output”])

5.4 性能优化与成本控制

现象：Agent响应慢，或者使用昂贵LLM（如GPT-4）时API调用成本激增。根因：每次工具调用和反思都是一次LLM API请求，复杂的任务会导致多次往返。解决方案：

工具设计聚合化：如果一个复杂操作需要多次调用LLM，考虑能否设计一个更强大的工具，在工具内部完成复杂逻辑，减少Agent的“思考-行动”轮次。
使用更便宜的LLM进行规划：可以采用“双LLM”策略。用一个快速、便宜的模型（如GPT-3.5 Turbo）负责规划和工具选择，再用一个强大、昂贵的模型（如GPT-4）负责最终答案的润色和总结。这需要更复杂的架构设计。
设置超时和最大迭代次数：使用max_execution_time和max_iterations参数严格限制Agent的单次运行时长和思考步数，防止因复杂或无法解决的问题而产生无限循环和巨额费用。
缓存（Caching）：对于重复性查询，特别是SQL Agent中描述数据库Schema的步骤，可以使用LangChain的缓存组件（如SQLiteCache）来缓存LLM的响应，显著提升速度并降低成本。

构建稳定、高效、安全的Agent系统是一个持续迭代的过程。从设计好一个单一功能的Tool开始，到组装成能协同工作的Toolkit，再到赋予其思考能力的Python Agent，最后到领域专家SQL Agent，每一步都考验着我们对问题拆解、工具抽象和提示词工程的理解。最重要的是，始终把安全和控制放在第一位，让AI在划定的边界内为我们创造价值。