Hello-Agents 《从零开始构建智能体》实践

她在丛中笑 發表於 2025-11-22 22:52:00

Hello-Agents 《从零开始构建智能体》实践

最近在DataWhale参加Agents开发的组队学习，虽然课程才刚满两周，但我这颗想搞事的心早就按捺不住了！今天，我终于要正式“出道”，成为一名智能体系统构建者啦～ 我最近调研发现某部门的小伙伴们得频繁巡检机房设备，工作量巨大不说，机房那噪音、辐射、还有不明气体……简直是对健康的“全方位关爱” 😅 于是他们悄悄问我：能不能让AI来帮忙巡检？我一听，这不就是我学《从零开始构建智能体》之后练手的好机会嘛！灵感瞬间爆发，火速整理思路，项目火速上线👇
 
项目名称：机房巡检助手
问题分析：当前机房巡检工作面临几个核心痛点：人工巡检效率低下，工作环境对人员不友好，机房内的噪音、辐射和有害气体影响健康。传统巡检模式存在盲区，易出错等问题。
核心功能：通过API工具采集机房环境数据，智能分析与决策模块负责数据处理与决策。实现根因分析：当多指标异常时，能智能推断故障根源，辅助快速定位问题。人机交互支持聊天对话。 
预期成果：实现聊天对话框内巡检机房环境数据，并智能分析总结，辅助快速定位问题。
 项目地址：https://gitee.com/yisheng163/hello-agents-ys 由于从 GitHub 拉取/推送代码的网络延迟较高，为提升协作效率，我将项目同步开源至 Gitee。
 效果图：
<img src="https://img2024.cnblogs.com/blog/27928/202511/27928-20251123000455000-2089498429.png">
 
 
<h4>开发环境准备</h4>
1，下载 vs code 安装。
2，# 安装HelloAgents
pip install hello-agents
测试 python -c "import hello_agents; print(dir(hello_agents))"
安装gradio库，低代码搭建聊天对话界面。
pip install gradio 编写代码
<div class="cnblogs_code"><img id="code_img_closed_fb15004b-2dce-453a-9f42-bead1383fafe" class="code_img_closed lazyload" data-src="http://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif"><img id="code_img_opened_fb15004b-2dce-453a-9f42-bead1383fafe" class="code_img_opened lazyload" style="display: none" data-src="http://images.cnblogs.com/OutliningIndicators/ExpandedBlockStart.gif">
<div id="cnblogs_code_open_fb15004b-2dce-453a-9f42-bead1383fafe" class="cnblogs_code_hide">
<pre>import os
from openai import OpenAI
from dotenv import load_dotenv
from typing import List, Dict

# 加载 .env 文件中的环境变量
load_dotenv()

class HelloAgentsLLM:
"""
为本书 "Hello Agents" 定制的LLM客户端。
它用于调用任何兼容OpenAI接口的服务，并默认使用流式响应。
"""
def __init__(self, model: str = None, apiKey: str = None, baseUrl: str = None, timeout: int = None):
 """
 初始化客户端。优先使用传入参数，如果未提供，则从环境变量加载。
 """
 self.model = model or os.getenv("LLM_MODEL_ID")
 apiKey = apiKey or os.getenv("LLM_API_KEY")
 baseUrl = baseUrl or os.getenv("LLM_BASE_URL")
 timeout = timeout or int(os.getenv("LLM_TIMEOUT", 60))

 if not all():
 raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。")

 self.client = OpenAI(api_key=apiKey, base_url=baseUrl, timeout=timeout)

def think(self, messages: List], temperature: float = 0) -> str:
 """
 调用大语言模型进行思考，并返回其响应。
 """
 print(f"🧠 正在调用 {self.model} 模型...")
 try:
 response = self.client.chat.completions.create(
 model=self.model,
 messages=messages,
 temperature=temperature,
 stream=True,
 )

 # 处理流式响应
 print("✅ 大语言模型响应成功:")
 collected_content = []# 创建空列表用于收集所有响应片段
 for chunk in response:# 遍历流式响应中的每个数据块
 content = chunk.choices.delta.content or ""# 提取当前数据块中的文本内容，如果没有则为空字符串

 print(content, end="", flush=True)# 实时打印内容，不换行并立即刷新输出缓冲区
 collected_content.append(content)# 将当前内容添加到收集列表中
 print()# 在流式输出结束后换行
 return "".join(collected_content)# 将收集的所有内容拼接成完整字符串并返回

 except Exception as e:
 print(f"❌ 调用LLM API时发生错误: {e}")
 return None

def invoke(self, messages: List], **kwargs) -> str:
 """
 兼容SimpleAgent的调用方法
 """
 temperature = kwargs.get("temperature", 0)
 return self.think(messages, temperature)

# --- 客户端使用示例 ---
if __name__ == '__main__':
try:
 llmClient = HelloAgentsLLM()

 exampleMessages = [
 {"role": "system", "content": "You are a helpful assistant that writes Python code."},
 {"role": "user", "content": "写一个快速排序算法"}
 ]

 print("--- 调用LLM ---")
 responseText = llmClient.think(exampleMessages)
 if responseText:
 print("\n\n--- 完整模型响应 ---")
 print(responseText)

except ValueError as e:
 print(e)</pre>
</div>
HelloAgentsLLM.py</div>
<div class="cnblogs_code"><img id="code_img_closed_4808bc8d-4722-46ff-8498-d6b5fa42ce7c" class="code_img_closed lazyload" data-src="http://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif"><img id="code_img_opened_4808bc8d-4722-46ff-8498-d6b5fa42ce7c" class="code_img_opened lazyload" style="display: none" data-src="http://images.cnblogs.com/OutliningIndicators/ExpandedBlockStart.gif">
<div id="cnblogs_code_open_4808bc8d-4722-46ff-8498-d6b5fa42ce7c" class="cnblogs_code_hide">
<pre># my_llm.py
import os
from typing import Optional
from openai import OpenAI
from hello_agents import HelloAgentsLLM

class MyLLM(HelloAgentsLLM):
def __init__(
 self,
 model: Optional = None,
 api_key: Optional = None,
 base_url: Optional = None,
 provider: Optional = "auto",
 **kwargs
):
 # 检查provider是否为我们想处理的'modelscope'
 if provider == "modelscope":
 print("正在使用自定义的 ModelScope Provider")
 self.provider = "modelscope"

 # 解析 ModelScope 的凭证
 self.api_key = api_key or os.getenv("MODELSCOPE_API_KEY")
 self.base_url = base_url or "https://api-inference.modelscope.cn/v1/"

 # 验证凭证是否存在
 if not self.api_key:
 raise ValueError("ModelScope API key not found. Please set MODELSCOPE_API_KEY environment variable.")

 # 设置默认模型和其他参数
 self.model = model or os.getenv("LLM_MODEL_ID") or "Qwen/Qwen2.5-VL-72B-Instruct"
 self.temperature = kwargs.get('temperature', 0.7)
 self.max_tokens = kwargs.get('max_tokens')
 self.timeout = kwargs.get('timeout', 60)

 # 使用获取的参数创建OpenAI客户端实例
 self._client = OpenAI(api_key=self.api_key, base_url=self.base_url, timeout=self.timeout)

 else:
 # 如果不是 modelscope, 则完全使用父类的原始逻辑来处理
 super().__init__(model=model, api_key=api_key, base_url=base_url, provider=provider, **kwargs)</pre>
</div>
my_llm.py</div>
配置.env文件 编写通过API查询机房环境信息的类
<div class="cnblogs_code"><img id="code_img_closed_f4df6049-7784-43f7-81b9-2b9e0a9ad090" class="code_img_closed lazyload" data-src="http://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif"><img id="code_img_opened_f4df6049-7784-43f7-81b9-2b9e0a9ad090" class="code_img_opened lazyload" style="display: none" data-src="http://images.cnblogs.com/OutliningIndicators/ExpandedBlockStart.gif">
<div id="cnblogs_code_open_f4df6049-7784-43f7-81b9-2b9e0a9ad090" class="cnblogs_code_hide">
<pre>import os
import requests
import hashlib
import json
from typing import Optional, Union

class api_Temperature:
def __init__(self):
 self.session = requests.Session()
 self.base_url = "http://192.169.200.5:3900"
 self.base_url = os.getenv("ROOM_API_URL")

 # 获取加盐值
 salt_url = f"{self.base_url}/api/Plugin.Base/Log/Salt"

 try:
 response = self.session.get(salt_url)
 response.raise_for_status()
 re_body = response.json()
 # print(f"re_body: {re_body}")

 fixed_salt = re_body.get("data", {}).get("FixedSalt")
 temp_salt = re_body.get("data", {}).get("TempSalt")

 # print(f"fixed_salt: {fixed_salt}")
 # print(f"temp_salt: {temp_salt}")

 # 登录
 login_url = f"{self.base_url}/api/Plugin.Base/Log/In"

 account = os.getenv("ROOM_API_USERNAME")
 password = os.getenv("ROOM_API_USERPASS")
 remember_me = "true"

 # 计算密文 md5(md5({账号}+{固定加盐值}+{密码})+{临时加盐值})
 password_step1 = self.md5_hash(account + fixed_salt + password)
 encrypted_password = self.md5_hash(password_step1 + temp_salt)

 post_data = {
 "Account": account,
 "Password": encrypted_password,
 "RememberMe": remember_me
 }

 login_response = self.session.post(login_url, data=post_data)
 login_response.raise_for_status()

 # 输出登陆结果
 print("=" * 50)
 # print("登录结果详情:")
 # print(f"请求URL: {login_url}")
 # print(f"请求方法: POST")
 # print(f"请求体: {login_response.request.body}")
 print(f"响应状态码: {login_response.status_code}")
 print(f"响应内容: {login_response.text}")
 print("=" * 50)

 except requests.exceptions.RequestException as e:
 print(f"初始化失败: {e}")

@staticmethod
def md5_hash(text: str) -> str:
 """MD5哈希计算"""
 return hashlib.md5(text.encode('utf-8')).hexdigest()

def get_temperature_value(self, object_id: str) -> float:
 """取单个温度值"""
 re_value = 0.0
 url = f"{self.base_url}/api/Plugin.Data/Property/RealTimeData"

 params = {
 "Clazz": "Plugin.Env.Model.Device",
 "objectId": object_id,
 "provider": "Plugin.Env.Providers.Data.Device.Property_Point",
 "propertyId": "HT_Temp"
 }

 try:
 response = self.session.get(url, params=params)
 response.raise_for_status()
 re_body = response.json()

 # 屏幕打印日志
 #print(f"[{self.get_current_time()}] {json.dumps(re_body)}")

 ht_temp_value = re_body.get("data", {}).get("value")
 if ht_temp_value:
 re_value = float(ht_temp_value)

 except (requests.exceptions.RequestException, ValueError) as e:
 print(f"获取温度值失败: {e}")

 return re_value

def get_leak_judge_value(self, object_id: str, property_id: str) -> bool:
 """取单个漏水判定"""
 url = f"{self.base_url}/api/Plugin.Data/Property/RealTimeData"

 params = {
 "Clazz": "Plugin.Env.Model.Device",
 "objectId": object_id,
 "provider": "Plugin.Env.Providers.Data.Device.Property_Point",
 "propertyId": property_id
 }

 try:
 response = self.session.get(url, params=params)
 response.raise_for_status()
 re_body = response.json()
 #print(json.dumps(re_body))

 mk_judge_value = re_body.get("data", {}).get("value")
 re_value = float(mk_judge_value)
 return re_value == 1

 except (requests.exceptions.RequestException, ValueError) as e:
 print(f"获取漏水判定失败: {e}")
 return False

def get_leak_locate_value(self, object_id: str, property_id: str) -> float:
 """取单个漏水位置"""
 re_value = 0.0
 url = f"{self.base_url}/api/Plugin.Data/Property/RealTimeData"

 params = {
 "Clazz": "Plugin.Env.Model.Device",
 "objectId": object_id,
 "provider": "Plugin.Env.Providers.Data.Device.Property_Point",
 "propertyId": property_id
 }

 try:
 response = self.session.get(url, params=params)
 response.raise_for_status()
 re_body = response.json()
 #print(json.dumps(re_body))

 mk_locate_value = re_body.get("data", {}).get("value")
 re_value = float(mk_locate_value)

 except (requests.exceptions.RequestException, ValueError) as e:
 print(f"获取漏水位置失败: {e}")

 return re_value

def get_all_value(self) -> str:
 """获取所有值"""

 # 分院信息
 SD_Temp_Main = 0.0
 SD_Temp_UPS = 0.0

 # 获取温度值
 SD_Temp_Main = self.get_temperature_value("30.wsd5")
 SD_Temp_UPS = self.get_temperature_value("30.wsd1")

 # 获取漏水判断值
 SD_Leak_Judge_Main1 = self.get_leak_judge_value("30.ls1", "Leak_Judge")
 SD_Leak_Judge_Main2 = self.get_leak_judge_value("30.ls2", "Leak_Judge")

 SD_Leak_Locate_Main1 = 0.0
 SD_Leak_Locate_Main2 = 0.0

 # 处理漏水定位逻辑(@ref)
 if SD_Leak_Judge_Main1:
 SD_Leak_Locate_Main1 = self.get_leak_locate_value("30.ls1", "Leak_Locate")

 if SD_Leak_Judge_Main2:
 SD_Leak_Locate_Main2 = self.get_leak_locate_value("30.ls2", "Leak_Locate")
 if SD_Leak_Locate_Main2 > SD_Leak_Locate_Main1:
 SD_Leak_Locate_Main1 = SD_Leak_Locate_Main2

 #获取总院信息

 LY_Temp_Main = 0.0
 LY_Temp_UPS = 0.0

 # 获取温度值(@ref)
 LY_Temp_Main = self.get_temperature_value("32.wsd3")
 LY_Temp_UPS = self.get_temperature_value("32.wsd1")

 # 获取漏水判断值
 LY_Leak_Judge_Main = self.get_leak_judge_value("32.mk", "Path0")
 LY_Leak_Judge_UPS = self.get_leak_judge_value("32.mk", "Path1")

 # 构建分院报告
 msgBody = "分院院区"
 if SD_Temp_Main == 0:
 msgBody += "，中心机房温度探测器故障"
 if SD_Temp_UPS == 0:
 msgBody += "，电源机房温度探测器故障"
 if SD_Temp_Main > 0:
 msgBody += f"，中心机柜温度{SD_Temp_Main}"
 if SD_Temp_UPS > 0:
 msgBody += f"，电源机柜温度{SD_Temp_UPS}"

 # 漏水状态判断

 if not SD_Leak_Judge_Main1 and not SD_Leak_Judge_Main2:
 msgBody += "，中心机柜无漏水"
 if SD_Leak_Judge_Main1 or SD_Leak_Judge_Main2:
 msgBody += f"，中心机柜漏水，漏水位置传感绳第{SD_Leak_Locate_Main1}米处"

 msgBody += "。"

 msgBody += "\n总院院区"

 # 构建总院院区报告
 if LY_Temp_Main == 0:
 msgBody += "，中心机柜温度探测器故障"
 if LY_Temp_UPS == 0:
 msgBody += "，电源机柜温度探测器故障"
 if LY_Temp_Main > 0:
 msgBody += f"，中心机柜温度{LY_Temp_Main}"
 if LY_Temp_UPS > 0:
 msgBody += f"，电源机柜温度{LY_Temp_UPS}"

 # 漏水状态判断
 if not LY_Leak_Judge_Main:
 msgBody += "，中心机柜无漏水"
 if LY_Leak_Judge_Main:
 msgBody += "，中心机柜漏水"
 if not LY_Leak_Judge_UPS:
 msgBody += "，电源机柜无漏水"
 if LY_Leak_Judge_UPS:
 msgBody += "，电源机柜漏水"

 msgBody += "。"

 return msgBody

if __name__ == "__main__":
# 创建API实例（会自动登录）
api = api_Temperature()

# 获取所有值
getall = api.get_all_value()
print(getall)</pre>
</div>
api_Temperature.py</div>
<div>创建包含机房巡检的工具注册表类</div>
<div>
<div class="cnblogs_code"><img id="code_img_closed_830170c5-4dd3-44fe-85a8-a2e343b2d4b1" class="code_img_closed lazyload" data-src="http://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif"><img id="code_img_opened_830170c5-4dd3-44fe-85a8-a2e343b2d4b1" class="code_img_opened lazyload" style="display: none" data-src="http://images.cnblogs.com/OutliningIndicators/ExpandedBlockStart.gif">
<div id="cnblogs_code_open_830170c5-4dd3-44fe-85a8-a2e343b2d4b1" class="cnblogs_code_hide">
<pre># my_calculator_tool2.py# 导入所需的模块和库
import ast # 抽象语法树模块，用于解析表达式
import operator # 操作符模块，提供基本的数学操作函数
import math # 数学模块，提供数学函数
from hello_agents import ToolRegistry# 从hello_agents导入ToolRegistry类
from api_Temperature import api_Temperature #取机房数据

def my_room_temp(expression: str) -> str:
"""机房巡检"""
if not expression.strip(): # 检查表达式是否为空
 return "巡项项目不能为空" # 如果为空则返回错误信息

try:
 if "机房" in expression or "温度" in expression:
 room_api = api_Temperature()
 msgBody = room_api.get_all_value()
 # 确保返回有效信息
 if msgBody:
 return msgBody
 else:
 return "未能获取到机房信息"
 if "漏水" in expression:
 room_api = api_Temperature()
 msgBody = room_api.get_all_value()
 # 确保返回有效信息
 if msgBody:
 return msgBody
 else:
 return "未能获取到漏水信息"

except Exception as e:
 return f"查询失败: {str(e)}" # 捕获异常并返回错误信息

def _eval_node(node, operators, functions):
"""简化的表达式求值"""
if isinstance(node, ast.Constant): # 如果节点是常量类型
 return node.value # 直接返回常量值
elif isinstance(node, ast.BinOp): # 如果节点是二元运算类型
 left = _eval_node(node.left, operators, functions) # 递归计算左子树
 right = _eval_node(node.right, operators, functions) # 递归计算右子树
 op = operators.get(type(node.op)) # 获取对应的操作符函数
 return op(left, right) # 执行运算并返回结果
elif isinstance(node, ast.Call): # 如果节点是函数调用类型
 func_name = node.func.id # 获取函数名
 if func_name in functions: # 检查函数是否受支持
 args = # 递归计算参数值
 return functions(*args) # 调用函数并返回结果
elif isinstance(node, ast.Name): # 如果节点是名称类型（如变量或常量）
 if node.id in functions: # 检查名称是否在函数列表中
 return functions # 返回对应的值

def create_calculator_registry():
"""创建包含机房巡检的工具注册表"""
registry = ToolRegistry() # 创建ToolRegistry实例

# 注册计算器函数
 registry.register_function(
 name="my_room_temp", # 工具名称
 description="机房巡检工具，支持温度，漏水检测",# 工具描述
 func=my_room_temp # 对应的函数
 )

return registry # 返回注册表实例

# 使用示例
if __name__ == "__main__":
msgBody=my_room_temp("机房巡检")
print(msgBody)</pre>
</div>
my_room_temp_tool.py</div>
主界面，通过gradio库，实现聊天对话界面。
<div class="cnblogs_code"><img id="code_img_closed_ac02633f-b3de-4fa1-b2cc-9300facd0c64" class="code_img_closed lazyload" data-src="http://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif"><img id="code_img_opened_ac02633f-b3de-4fa1-b2cc-9300facd0c64" class="code_img_opened lazyload" style="display: none" data-src="http://images.cnblogs.com/OutliningIndicators/ExpandedBlockStart.gif">
<div id="cnblogs_code_open_ac02633f-b3de-4fa1-b2cc-9300facd0c64" class="cnblogs_code_hide">
<pre>import gradio as gr
from hello_agents import SimpleAgent
from HelloAgentsLLM import HelloAgentsLLM
from my_room_temp_tool import create_calculator_registry

# 创建LLM实例
llm = HelloAgentsLLM()

# 创建工具注册表
registry = create_calculator_registry()

# 创建智能体
agent = SimpleAgent(
name="机房巡检助手",
llm=llm,
system_prompt="你是一个机房巡检助手，专门负责监控和报告机房的温度及漏水情况。当用户询问机房相关信息时，你应该使用my_room_temp工具来获取实时数据。"
)

# 重写agent的run方法，使其能够调用我们的工具
def agent_run_with_tools(input_text):
# 检查是否需要调用工具
if "机房" in input_text or "温度" in input_text or "漏水" in input_text:
 # 调用工具获取数据
 tool_result = registry.execute_tool("my_room_temp", input_text)
 # 构造最终回复
 messages = [
 {"role": "user", "content": f"根据以下机房巡检结果，用自然语言回答用户问题：{input_text}\n\n巡检结果：{tool_result}"}
 ]
 response = llm.think(messages)
 return response
else:
 # 直接调用agent.run
 return agent.run(input_text)

def agent_chat(message, history):
"""
与agent进行对话
"""
try:
 # 运行agent
 response = agent_run_with_tools(message)
 # 确保返回的是字符串而不是None
 return response or "抱歉，我没有收到有效的回复。"
except Exception as e:
 return f"发生错误: {str(e)}"

# 创建Gradio界面
with gr.Blocks(title="机房巡检助手") as demo:
gr.Markdown("# 机房巡检助手-- 《Hello-Agents从零开始构建智能体》之毕业设计")
gr.Markdown("这是一个聊天界面，可以与机房巡检助手进行交互，并自动调用my_room_temp工具。")

chatbot = gr.Chatbot(label="对话历史", type="messages")
msg = gr.Textbox(label="输入消息", placeholder="请输入您的问题，例如：机房温度是多少？")
clear = gr.Button("清除对话")
designer = gr.Markdown("设计：西湖谊生")

def add_user_message(history, message):
 """添加用户消息到历史记录"""
 return history + [{"role": "user", "content": message}]

def bot_response(history):
 """生成机器人回复并添加到历史记录"""
 user_message = history[-1]["content"]
 bot_reply = agent_chat(user_message, history)
 history.append({"role": "assistant", "content": bot_reply})
 return history, ""

msg.submit(add_user_message, , ).then(
 bot_response, ,
)
clear.click(lambda: [], None, chatbot, queue=False)

if __name__ == "__main__":
demo.launch(server_name="0.0.0.0", server_port=7861, share=False)</pre>
</div>
app.py</div>
  编写README.md，将代码提交到远程仓库。
 

</div> 
来源：https://www.cnblogs.com/yisheng163/p/19258818

頁: [1]

琼殿技术论坛's Archiver

Hello-Agents 《从零开始构建智能体》实践