吃透 LangChain Tool + Agent:从零实现可商用电商客服智能体
一、前言:为什么学完 RAG 必须学 Agent?
前面我们掌握了文档解析、向量检索、RAG 知识库问答,但 RAG 只能回答静态文档知识。真实企业场景需要:查订单、查库存、算价格、查售后规则——这些动态实时数据,必须依靠 Tool 工具调用 + Agent 智能体实现。本文带你从零掌握:Tool 原理、工具规范、参数校验、Agent 运行机制,最终手写一套可直接落地的电商客服智能体。
二、核心认知:什么是 Tool?为什么大模型需要工具调用?
2.1 大模型的天生短板
- 不懂实时业务数据(数据库/订单/库存)
- 不会计算、不会查询外部接口
- 只能基于训练知识和 Prompt 回答,无法“动手做事”
2.2 Tool 的核心作用
将 Python 函数封装为大模型可识别、可主动调用的外部能力。标准流程:用户提问 → 模型思考 → 调用工具 → 获取真实数据 → 生成答案
2.3 普通函数 VS LangChain Tool(核心区别)
Tool 相比普通函数,多了三套供模型识别的元数据:工具名、参数结构、功能描述,是模型能否正确调用的关键。
三、LangChain Tool 从零实战(由浅入深)
3.1 基础单参数 Tool 开发
@tool 装饰器用法:最简单的转换方式。
from langchain.tools import tool
模拟订单数据库
ORDER_DB = {
"1001": {"status": "已发货", "product": "《Python编程从入门到实践》"},
"1002": {"status": "待付款", "product": "无线蓝牙耳机"},
}
@tool
def get_order_status(order_id: str) -> str:
"""
根据订单ID查询订单的当前状态。
参数:
order_id (str): 订单的唯一标识符。
返回:
str: 订单状态的描述,例如“已发货”、“待付款”。如果订单不存在,返回“未找到该订单”。
"""
order = ORDER_DB.get(order_id)
if order:
return f"订单 {order_id} ({order['product']}) 状态为:{order['status']}"
else:
return f"未找到订单ID为 {order_id} 的记录。"
手动 invoke 测试(工程必备:先测工具再给 Agent 用):
# 测试正常查询
result = get_order_status.invoke({"order_id": "1001"})
print(result) # 输出:订单 1001 (《Python编程从入门到实践》) 状态为:已发货
测试异常查询
result = get_order_status.invoke({"order_id": "9999"})
print(result) # 输出:未找到订单ID为 9999 的记录。
查看工具元信息:name / description / args
print(f"工具名称: {get_order_status.name}")
print(f"工具描述: {get_order_status.description}")
print(f"工具参数: {get_order_status.args}")
3.2 多参数 Tool 实战
折扣价格计算工具,掌握多参数传递、类型规范、返回值标准化。
@tool
def calculate_discount_price(original_price: float, discount_rate: float = 0.0, coupon_amount: float = 0.0) -> float:
"""
计算商品折后价格。
参数:
original_price (float): 商品原价,必须大于0。
discount_rate (float): 折扣率,范围0-1(例如0.1代表9折)。默认为0。
coupon_amount (float): 优惠券抵扣金额。默认为0。
返回:
float: 折后最终价格。计算方式:原价 * (1 - 折扣率) - 优惠券金额。结果不会低于0。
"""
if original_price <= 0:
return 0.0
final_price = original_price * (1 - discount_rate) - coupon_amount
return max(final_price, 0.0)
3.3 企业级进阶:Pydantic 参数校验
解决模型传参不规范、参数错误问题:自定义参数 Schema、Field 字段描述、正则校验、自动拦截非法参数,避免工具报错。
from langchain.tools import tool
from pydantic import BaseModel, Field, field_validator
import re
1. 定义参数的数据模型
class QueryInventoryInput(BaseModel):
product_sku: str = Field(
...,
description="商品的SKU编码,格式必须为'PROD-XXX',其中XXX为3位数字。",
examples=["PROD-001", "PROD-123"]
)
warehouse_id: str = Field(
"WH01",
description="仓库编号,默认为'WH01'。",
examples=["WH01", "WH02"]
)
# 2. 使用验证器进行高级校验
@field_validator('product_sku')
@classmethod
def validate_sku_format(cls, v):
pattern = r'^PROD-\d{3}$'
if not re.match(pattern, v):
raise ValueError(f"SKU格式错误,必须符合 {pattern}")
return v
3. 绑定 args_schema
@tool(args_schema=QueryInventoryInput)
def query_inventory(product_sku: str, warehouse_id: str = "WH01") -> str:
"""
根据商品SKU和仓库编号查询实时库存数量。
"""
# 模拟库存数据
inventory_db = {
("PROD-001", "WH01"): 15,
("PROD-001", "WH02"): 8,
("PROD-123", "WH01"): 0,
}
key = (product_sku, warehouse_id)
stock = inventory_db.get(key, 0)
return f"商品 {product_sku} 在仓库 {warehouse_id} 的库存为:{stock} 件。"
3.4 Tool 编写黄金规范(避坑重点)
- 描述必须清晰:用途、场景、参数、返回值
- 单一职责:一个工具只做一件事
- 命名规范:小写下划线
- 返回结果清晰可读,禁止返回 ok/true 等模糊内容
四、Agent 核心原理:真正让 AI 自主干活
4.1 Agent 是什么?和 Chain 的本质区别
Chain:流程固定,无脑执行。Agent:自主思考、判断、选工具、循环执行。
4.2 核心公式:Agent = Model + Tool + Harness
通俗解读:
- Model:大脑,负责思考决策
- Tool:能力,负责具体干活
- Harness:调度管家,负责循环、调用、传参、收尾
4.3 Agent 完整运行链路
用户问题 → 模型判断是否需要工具 → 自动选择工具 → 执行工具 → 拿到结果 → 最终回答
五、从零搭建第一个智能 Agent(代码实操)
5.1 环境准备
依赖安装、DeepSeek 模型配置、env 密钥管理:
# 所需依赖包
pip install langchain langchain-openai python-dotenv
使用国内清华镜像加速安装
pip install langchain langchain-openai python-dotenv -i https://pypi.tuna.tsinghua.edu.cn/simple
# .env 文件内容
DEEPSEEK_API_KEY=your_deepseek_api_key_here
5.2 单工具 Agent 快速搭建
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import tool
from langchain_core.prompts import ChatPromptTemplate
load_dotenv()
1. 定义工具(复用前面的订单查询工具)
@tool
def get_order_status(order_id: str) -> str:
"""根据订单ID查询订单状态。"""
ORDER_DB = {"1001": "已发货", "1002": "待付款"}
status = ORDER_DB.get(order_id, "订单不存在")
return f"订单 {order_id} 状态: {status}"
2. 初始化大模型(以DeepSeek为例)
model = ChatOpenAI(
model="deepseek-chat",
openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com"
)
3. 准备工具列表
tools = [get_order_status]
4. 创建Agent提示词模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的电商客服助手。请根据用户问题,判断是否需要使用工具查询订单状态。如果需要,请调用工具。请用中文回答。"),
("placeholder", "{chat_history}"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
5. 使用create_react_agent创建智能体
agent = create_react_agent(model, tools, prompt)
6. 创建Agent执行器
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)
7. 执行测试
if name == "main":
question = "帮我查一下订单1001的状态"
result = agent_executor.invoke({"input": question})
print("\n=== 最终回答 ===")
print(result["output"])
5.3 查看 Agent 完整消息流
可视化展示:用户消息 → AI 工具调用 → 工具返回 → AI 最终回答,彻底看懂 Agent 执行逻辑。
# 查看完整的执行消息流
for message in result.get("intermediate_steps", []):
print(f"步骤: {message}")
5.4 多工具协同 Agent
一个 Agent 挂载多个工具,实现自动根据用户问题智能选工具。
# 定义多个工具
tools = [get_order_status, query_inventory, calculate_discount_price]
更新提示词,告诉 Agent 有多个工具可用
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的电商客服助手,可以处理订单查询、库存查询和价格计算。请根据用户问题自动选择合适的工具。如果问题需要多个信息,可以依次调用多个工具。请用中文回答。"),
("placeholder", "{chat_history}"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
agent = create_react_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)
测试多工具协同
questions = [
"我的订单1001发货了吗?",
"商品PROD-001在WH01的库存是多少?",
"原价1000元打8折后多少钱?",
"订单1002的状态和库存PROD-001的情况。",
]
for q in questions:
result = agent_executor.invoke({"input": q})
print(f"问题: {q}")
print(f"回答: {result['output']}")
六、企业级实战:完整电商客服 Agent 项目
6.1 项目需求与目录结构
实现:订单查询、库存查询、价格计算、退款规则咨询
ecommerce_agent/
├── tools.py # 所有业务工具实现
├── customer_agent.py # 客服Agent封装
├── main.py # 交互式终端程序
├── .env # 环境变量配置
└── requirements.txt # 依赖包列表
6.2 统一封装业务工具 tools.py
"""电商客服工具集"""
from langchain.tools import tool
from pydantic import BaseModel, Field
from typing import Optional
ORDER_DATABASE = {
"ORD-2024-001": {"status": "已发货", "product": "iPhone 15 Pro", "amount": 8999.00},
"ORD-2024-002": {"status": "待付款", "product": "MacBook Air", "amount": 8499.00},
"ORD-2024-003": {"status": "已完成", "product": "AirPods Pro", "amount": 1899.00},
}
INVENTORY_DATABASE = {
"iPhone 15 Pro": {"stock": 15, "price": 8999.00},
"MacBook Air": {"stock": 8, "price": 8499.00},
"AirPods Pro": {"stock": 0, "price": 1899.00},
"iPad Air": {"stock": 22, "price": 4799.00},
}
REFUND_POLICY = {
"7天无理由退货": "签收后7天内,商品完好、包装齐全,可申请无理由退货",
"质量问题退货": "商品存在质量问题,30天内可申请退货",
"退款到账时间": "审核通过后,3-7个工作日内原路退回",
}
@tool
def get_order_details(order_id: str) -> str:
"""根据订单ID查询订单详细信息。"""
order = ORDER_DATABASE.get(order_id)
if order:
return f"订单 {order_id}: 商品[{order['product']}],状态[{order['status']}],金额[{order['amount']}元]"
return f"未找到订单 {order_id},请确认订单编号是否正确。"
class InventoryQueryInput(BaseModel):
product_name: str = Field(..., description="商品名称,如iPhone 15 Pro")
warehouse: Optional[str] = Field("默认仓库", description="仓库名称")
@tool(args_schema=InventoryQueryInput)
def check_inventory(product_name: str, warehouse: str = "默认仓库") -> str:
"""查询指定商品的库存情况。"""
product = INVENTORY_DATABASE.get(product_name)
if product:
stock_status = "有货" if product["stock"] > 0 else "缺货"
return f"{product_name} 在{warehouse}的库存: {product['stock']}件 ({stock_status}),价格: {product['price']}元"
return f"未找到商品 {product_name},请确认商品名称是否正确。"
@tool
def calculate_final_price(original_price: float, discount_percent: float = 0, coupon: float = 0) -> str:
"""计算商品的最终价格。"""
if original_price <= 0:
return "价格必须大于0"
discount_amount = original_price * (discount_percent / 100)
final_price = original_price - discount_amount - coupon
final_price = max(final_price, 0)
return f"原价{original_price}元,折扣{discount_percent}%减{discount_amount:.2f}元,优惠券减{coupon}元,最终价格: {final_price:.2f}元"
@tool
def get_refund_policy(policy_type: str = "全部") -> str:
"""查询电商平台的退款规则。"""
if policy_type == "全部":
result = "退款规则汇总:\n"
for key, value in REFUND_POLICY.items():
result += f"- {key}: {value}\n"
return result.strip()
policy = REFUND_POLICY.get(policy_type)
if policy:
return f"{policy_type}: {policy}"
return f"未找到'{policy_type}'对应的退款规则"
ALL_TOOLS = [get_order_details, check_inventory, calculate_final_price, get_refund_policy]
6.3 客服 Agent 封装
"""电商客服Agent封装"""
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from tools import ALL_TOOLS
def create_customer_agent(verbose: bool = False):
load_dotenv()
model = ChatOpenAI(
model="deepseek-chat",
openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com",
temperature=0.1,
)
system_prompt = """你是一个专业的电商客服助手。
你可以使用的工具:
get_order_details - 查询订单详情,需要订单编号
check_inventory - 查询商品库存,需要商品名称
calculate_final_price - 计算最终价格,需要原价、可选折扣和优惠券
get_refund_policy - 查询退款规则
重要规则:
只能使用上述工具获取信息,不能编造数据
如果用户问题缺少必要信息,请主动询问
回答时请基于工具返回的结果,不要自行补充或修改"""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("placeholder", "{chat_history}"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
agent = create_react_agent(model, ALL_TOOLS, prompt)
executor = AgentExecutor(agent=agent, tools=ALL_TOOLS, verbose=verbose, handle_parsing_errors=True)
return executor
6.4 交互式终端问答程序
"""终端客服主程序"""
from customer_agent import create_customer_agent
def main():
agent = create_customer_agent(verbose=False)
print("智能客服已启动,输入 'exit' 退出。")
while True:
user_input = input("\n用户: ")
if user_input.lower() == "exit":
print("再见!")
break
response = agent.invoke({"input": user_input})
print(f"客服: {response['output']}")
if name == "main":
main()
6.5 组合场景测试
复杂问题自动多轮调用工具,例如:查询订单状态 + 判断是否可以退款。
用户: 订单ORD-2024-001的状态是什么?如果还没完成,能不能退款?
客服: 订单ORD-2024-001的状态是“已发货”。根据退款规则,您可以在签收后7天内申请无理由退货,或30天内因质量问题退货。当前状态为已发货,建议您先收货检查商品状况。
七、工程化规范与落地经验
7.1 Tool 设计规范
- 查询类工具、写入类工具区分
- 高危操作必须加校验与确认
- 工具解耦、可复用、可单独测试
7.2 Agent 适用场景
- 适合:客服咨询、数据查询、办公助手、运维查询
- 不适合:固定流程、单一任务(用 Chain 更简单)
八、高频报错与避坑总结
- Agent 不调用工具:描述不清晰、prompt 约束不足
- 调错工具:工具功能重叠、语义区分度低
- 参数报错:未做 Pydantic 校验、模型传参不规范
- 数据虚假:未强制依赖工具返回结果
更多推荐


所有评论(0)