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

她在丛中笑

最近在DataWhale参加Agents开发的组队学习，虽然课程才刚满两周，但我这颗想搞事的心早就按捺不住了！今天，我终于要正式“出道”，成为一名智能体系统构建者啦～

我最近调研发现某部门的小伙伴们得频繁巡检机房设备，工作量巨大不说，机房那噪音、辐射、还有不明气体……简直是对健康的“全方位关爱” 😅
于是他们悄悄问我：能不能让AI来帮忙巡检？我一听，这不就是我学《从零开始构建智能体》之后练手的好机会嘛！灵感瞬间爆发，火速整理思路，项目火速上线👇

 

项目名称：机房巡检助手

问题分析：当前机房巡检工作面临几个核心痛点：人工巡检效率低下，工作环境对人员不友好，机房内的噪音、辐射和有害气体影响健康。传统巡检模式存在盲区，易出错等问题。

核心功能：通过API工具采集机房环境数据，智能分析与决策模块 负责数据处理与决策。实现根因分析：当多指标异常时，能智能推断故障根源，辅助快速定位问题。人机交互支持聊天对话。 

预期成果：实现聊天对话框内巡检机房环境数据，并智能分析总结，辅助快速定位问题。

项目地址：https://gitee.com/yisheng163/hello-agents-ys
由于从 GitHub 拉取/推送代码的网络延迟较高，为提升协作效率，我将项目同步开源至 Gitee。

 效果图：

 

 

开发环境准备

1，下载 vs code 安装 。

2，# 安装HelloAgents

pip install hello-agents

测试 python -c "import hello_agents; print(dir(hello_agents))"

安装gradio库，低代码搭建聊天对话界面。

pip install gradio

编写代码

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([self.model, apiKey, baseUrl]):
            raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。")

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

    def think(self, messages: List[Dict[str, str]], 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[0].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[Dict[str, str]], **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)

HelloAgentsLLM.py

# 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[str] = None,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        provider: Optional[str] = "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)

my_llm.py

配置.env文件

编写通过API查询机房环境信息的类

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
        
        # 处理漏水定位逻辑[1](@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
        
        # 获取温度值[2](@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)

api_Temperature.py

创建包含机房巡检的工具注册表类

# 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 = [_eval_node(arg, operators, functions) for arg in node.args]  # 递归计算参数值
            return functions[func_name](*args)       # 调用函数并返回结果
    elif isinstance(node, ast.Name):                # 如果节点是名称类型（如变量或常量）
        if node.id in functions:                    # 检查名称是否在函数列表中
            return functions[node.id]               # 返回对应的值

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)

my_room_temp_tool.py

主界面，通过gradio库，实现聊天对话界面。

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, [chatbot, msg], [chatbot]).then(
        bot_response, [chatbot], [chatbot, msg]
    )
    clear.click(lambda: [], None, chatbot, queue=False)

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7861, share=False)

app.py

 
编写README.md，将代码提交到远程仓库。

 

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