当前位置：首页 > news >正文

LangChain框架在高炉炼铁智能化领域的应用~系列文章09：工具调用Tool — 让AI学会操作高炉仪表盘

news 2026/6/16 22:06:01

🔧 第9期：工具调用Tool — 让AI学会操作高炉仪表盘

专栏：《LangChain框架在高炉炼铁智能化领域的应用》
前情回顾：上期我们构建了高炉Agent，有了"自主决策大脑"
本期重点：Tool —— Agent的"双手"，让AI真正"动手操作"

👐 引言：大脑再好，也得有手

上期我们把 Agent 比作"有自主意识的 AI 员工"。但你想过没有——

一个员工光有大脑，没有手，能干啥？🤷

🤖 Agent (大脑) ↓ 思考："我需要查一下当前风压数据" ↓ 决策："还需要查一下历史趋势对比" ↓ 结论："炉况有异常，需要报警" ↓ 但……手呢？谁来点鼠标？谁来按按钮？ 😱 卡住了！

这就是Tool（工具）的作用——它们是 Agent 的"双手"，让 AI 能真正操作外部系统！

人 = 大脑(思考) + 双手(操作) Agent = LLM(决策) + Tool(执行)

🛠️ LangChain 中的 Tool 体系

工具的分类

Tool 类型树： │ ├── 📡 数据获取类 —— 查参数、查数据库、查知识库 ├── 🔢 计算分析类 —— 数值计算、统计分析、公式推导 ├── ✉️ 通知通讯类 —— 发飞书、发邮件、发短信 ├── 🎮 控制操作类 —— 调整参数、下发指令 └── 🔌 外部API类 —— 调用第三方服务

高炉场景的工具清单

工具	类型	作用	Agent调用场景
`get_furnace_params`	数据获取	查实时参数	“看看当前炉况”
`get_param_history`	数据获取	查历史趋势	“铁温最近在下降吗”
`search_kb`	数据获取	搜知识库	“上次悬料怎么处理的”
`calculate_ti`	计算分析	算透气性指数	“帮我算下透气性”
`send_alert`	通知通讯	发飞书告警	“发现异常，通知值班”
`write_report`	数据输出	生成日报	“帮我生成今天的日报”

💻 进阶篇：从基础工具到高阶技巧

技巧1️⃣ 带参数校验的工具

fromlangchain.toolsimporttoolfrompydanticimportBaseModel,Field# 📌 定义工具的参数模型，自动做校验classFurnaceQueryInput(BaseModel):"""高炉参数查询的输入校验"""furnace_name:str=Field(description="高炉名称，如'5号高炉'、'6号高炉'",min_length=2,max_length=20)param_type:str=Field(description="参数类别：'temperature'(温度) / 'pressure'(压力) / 'all'(全部)",pattern="^(temperature|pressure|all)$")@tool(args_schema=FurnaceQueryInput)defquery_furnace_with_filter(furnace_name:str,param_type:str)->str:"""按类别筛选查询高炉参数"""all_params={"5号高炉":{"temperature":{"风温":"1205°C","铁水温度":"1490°C"},"pressure":{"风压":"0.392MPa","顶压":"0.22MPa"}}}furnace_data=all_params.get(furnace_name,{})ifparam_type=="all":result={}forvinfurnace_data.values():result.update(v)returnstr(result)returnstr(furnace_data.get(param_type,{}))# ✅ Agent调用时，如果参数不对会自动报错提示

技巧2️⃣ "组合式"工具

@tooldefcomprehensive_furnace_analysis(furnace_name:str)->str:""" 综合高炉炉况分析：一次调用完成参数查询+关键指标计算+初步诊断 furnace_name: 高炉名称 """# 1. 获取参数params=get_furnace_params_data(furnace_name)# 抽取公共逻辑# 2. 计算复合指标wind_volume=float(params["风量"].replace("m³/min",""))wind_pressure=float(params["风压"].replace("MPa",""))permeability=wind_volume/(wind_pressure*10000)# 透气性指数# 3. 初步诊断iron_temp=float(params["铁水温度"].replace("°C",""))ifiron_temp<1480:temp_status="⚠️ 偏低"elifiron_temp>1520:temp_status="⚠️ 偏高"else:temp_status="✅ 正常"returnf""" 【{furnace_name}综合诊断】 📊 关键参数：{params}🔢 透气性指数：{permeability:.1f}🌡️ 铁温状态：{temp_status}"""

技巧3️⃣ 带状态的上下文感知工具

fromdatetimeimportdatetime# 全局上下文（模拟）_global_context={"current_shift":"甲班","operator":"张工","last_alert_time":None}@tooldefsend_shift_alert(message:str,level:str="info")->str:""" 发送班次告警/通知 message: 告警内容 level: 告警级别 (info/warning/critical) """now=datetime.now()_global_context["last_alert_time"]=now urgency={"info":"📋","warning":"⚠️","critical":"🚨"}icon=urgency.get(level,"📋")alert=f"""{icon}【{level.upper()}】{now.strftime('%H:%M:%S')}👤 值班:{_global_context['current_shift']}-{_global_context['operator']}📝{message}"""# 真实场景这里会调用飞书/短信APIprint(f"🔔 告警已发送:{alert}")returnf"告警已发送：{message}"

🏭 实战：高炉智能体完整工具集

把前面所有技巧整合，打造高炉Agent的完整工具箱 🔧

# 📁 bf_toolkit.py# 高炉Agent完整工具集fromlangchain.toolsimporttoolfromlangchain.agents.middlewareimportwrap_tool_callfromlangchain.messagesimportToolMessagefromtypingimportOptional,ListfrompydanticimportBaseModel,Fieldimportjson# ─────────── 1. 数据获取类工具 ───────────@tooldefget_real_time_params(furnace_name:str)->str:""" 【高炉必用】获取高炉实时运行参数 furnace_name: 高炉名称编号，如"5号高炉" """# 模拟连接SCADA系统params={"5号高炉":{"timestamp":"2025-06-05 14:30:00","风温":"1205°C","风压":"0.392MPa","风量":"4920m³/min","铁水温度":"1490°C","硅含量":"0.42%","炉渣碱度":"1.18","透气性指数":"33.8","料速":"6.3批/h","顶温":"185°C","顶压":"0.22MPa"}}data=params.get(furnace_name)ifnotdata:returnf"❌ 未找到{furnace_name}的实时数据"returnjson.dumps(data,ensure_ascii=False,indent=2)@tooldefget_historical_data(furnace_name:str,param_name:str,hours:int=24,interval:str="1h")->str:""" 【趋势分析】查询参数的历史变化趋势 furnace_name: 高炉名称 param_name: 参数名称 hours: 回溯小时数，默认24 interval: 数据间隔，默认1h """# 模拟历史数据importrandom base_value={"铁水温度":1500,"硅含量":0.45,"风温":1200,"风压":0.39}base=base_value.get(param_name,100)trend=base-hours*0.5# 模拟下降趋势points=[]foriinrange(min(hours,24)):val=trend+i*0.3+random.uniform(-0.5,0.5)points.append({"time":f"{i:02d}:00","value":round(val,2)})returnjson.dumps({"furnace":furnace_name,"param":param_name,"trend":"下降"ifpoints[-1]["value"]<points[0]["value"]else"上升","data":points},ensure_ascii=False)# ─────────── 2. 计算分析类工具 ───────────@tooldefcalculate_comprehensive_index(wind_volume:float,wind_pressure:float,iron_temp:float,si_content:float)->str:""" 【专业计算】计算高炉综合评价指标 wind_volume: 风量 (m³/min) wind_pressure: 风压 (MPa) iron_temp: 铁水温度 (°C) si_content: 硅含量 (%) """# 透气性指数ti=wind_volume/(wind_pressure*10000)ifwind_pressure>0else0# 炉热指数heat_index=iron_temp*0.7+si_content*100*0.3# 综合评价score=0score+=30if1100<=iron_temp<=1250else10score+=30if0.35<=si_content<=0.55else10score+=20if30<=ti<=38else5score+=20if1480<=iron_temp<=1520else5returnjson.dumps({"透气性指数":round(ti,1),"炉热指数":round(heat_index,1),"综合评分":score,"评价":"优秀"ifscore>=80else"良好"ifscore>=60else"需关注","计算时间":"2025-06-05 14:30:00"},ensure_ascii=False)# ─────────── 3. 告警通知类工具 ───────────@tooldefsend_production_alert(furnace_name:str,alert_type:str,message:str,severity:str="warning")->str:""" 【告警通知】发送生产告警通知到中控室 furnace_name: 高炉名称 alert_type: 告警类型 (temperature/pressure/status) message: 告警内容 severity: 严重级别 (info/warning/critical) """alert_record={"timestamp":"2025-06-05 14:30:00","furnace":furnace_name,"type":alert_type,"severity":severity,"message":message,"status":"已发送"}# 模拟发送通知icons={"info":"📋","warning":"⚠️","critical":"🚨"}print(f"{icons.get(severity,'📋')}已向{furnace_name}中控室发送告警")returnjson.dumps(alert_record,ensure_ascii=False)# ─────────── 4. 工具错误处理 ───────────@wrap_tool_calldefbf_tool_middleware(request,handler):"""高炉工具的全局错误处理中间件"""try:result=handler(request)returnresultexceptValueErrorase:returnToolMessage(content=f"参数错误:{str(e)}，请检查输入参数格式",tool_call_id=request.tool_call["id"])exceptExceptionase:returnToolMessage(content=f"工具执行异常:{str(e)}，请稍后重试或联系系统管理员",tool_call_id=request.tool_call["id"])# ─────────── 5. 构建带完整工具集的Agent ───────────defbuild_full_agent():"""构建带完整工具集的高炉Agent"""fromlangchain.agentsimportcreate_agentfromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="doubao-seed-2-0-lite-260215",temperature=0.2,timeout=600)all_tools=[get_real_time_params,get_historical_data,calculate_comprehensive_index,send_production_alert]agent=create_agent(model=llm,system_prompt="""🏭 你是高炉炼铁智能体，拥有完整的高炉操作工具集。 【可用工具及使用场景】 1. get_real_time_params - 查实时参数（任何时候先查数据） 2. get_historical_data - 查历史趋势（分析问题时使用） 3. calculate_comprehensive_index - 计算综合评价指标 4. send_production_alert - 发送告警通知（发现异常时使用） 【工作流程】 1️⃣ 先获取实时数据 2️⃣ 如有需要查询历史趋势 3️⃣ 计算综合评价指标 4️⃣ 给出分析结论和建议 5️⃣ 如有异常及时发送告警 【安全规则】 - 告警需确认数据准确性后再发送 - critical级别告警必须附带数据依据""",tools=all_tools,middleware=[bf_tool_middleware])returnagent# ─────────── 运行测试 ───────────if__name__=="__main__":agent=build_full_agent()# 测试：完整的炉况分析+告警result=agent.invoke({"messages":[{"role":"user","content":"检查一下5号高炉炉况，铁水温度1490°C，硅含量0.38%，""风温1205°C，风压0.392MPa，风量4920m³/min。""帮我分析并给出建议，如果异常就发告警。"}]})print(result["messages"][-1].content)

⚡ Tool 开发黄金法则

法则1：单一职责

# ❌ 错误：一个工具干所有事@tooldefdo_everything(data:str)->str:...# ✅ 正确：每个工具只干一件事@tooldefget_temperature(furnace:str)->str:...# 只查温度@tooldefget_pressure(furnace:str)->str:...# 只查压力

法则2：详细的工具描述

# ❌ 抽象描述@tooldefget_data(name:str)->str:"""获取数据"""...# ✅ 具体描述（模型需要靠描述决定是否调用！）@tooldefget_furnace_params(name:str)->str:""" 当用户需要查询高炉的实时运行参数时使用此工具。 参数包括：风温、风压、风量、铁水温度、硅含量等。 适用于：判断炉况、查看指标、日常巡检。 name参数为高炉编号，如"5号高炉"。 """...

法则3：防呆设计

@tooldefquery_furnace(furnace_name:str)->str:"""查询高炉参数"""# 自动容错valid_names={"5号高炉","5#高炉","BF-5","5号炉"}iffurnace_nameinvalid_names:furnace_name="5号高炉"# 统一名称# ...

📊 本期小结

知识点	一句话总结
Tool是什么	Agent的"双手"，让AI能操作外部系统
参数校验	用 Pydantic 做输入校验，防止"乱传参"
组合工具	一次调用完成多步操作
错误处理	用中间件统一处理工具异常
开发法则	单一职责、详细描述、防呆设计

核心心法：