hi，大家好！在 @弈心 王印老师和 @朱嘉盛 朱老师的的带领下，我成为了《网络工程师的AI之路：LLM大语言模型》的首批读者，再次入局AI，努力从netdevops向AIops转型。如果你对这方面感兴趣，可以联系王老师本人进行交流并获取相关资料。
历经一个半月的钻研，如今也算是取得了一定成果（至少达到了自己的预期）。于是，我萌生了撰写系列文章，记录整个实践过程的想法。
本次构建的智能体主要借助 OpenAI Agents SDK、langchain、ragflow、netmiko 和 streamlit 得以实现，本文将详细阐述这些技术的应用。来来来，废话不多说，直接开始了！
以下是我的AI网络运维智能体的界面：
从图中的灰色小字也可以看到，目前实现了4种功能：
当用户反馈某个地点网络中断，请求协助排查时，AI 智能体会依照预设的排查流程，依次执行网络连通性检查、路由状态检测、提取流量监控、路由协议检查等一系列检查操作。它会综合所有检查结果，进行深度汇总分析，精准告知用户问题所在，并给出下一步排查建议。若检测出是广域网线路中断，AI 智能体将自动触发 5G 隧道开启程序，实现网络的自动修复，保障业务连续性。
如果发现是广域网线路中断，AI智能体会自动开启5G隧道，实现网络自动修复：
我所在的分公司IP地址资源有点紧张，技术人员常常因为哪个IP能用而头疼。所以增加了这个统计IP资源利用率和查询某个地点可用IP的功能。
用户只需提供 IP 地址，智能体就能利用网络拓扑信息以及设备管理协议，精准定位到该终端连接在哪台交换机的哪个接口，为网络设备管理与故障排查提供了便捷的定位手段。
例如，当询问 “iperf 命令怎样使用” 时，智能体首先会在 ragflow 知识库中进行精准检索，若检索到的答案不够详尽，它会凭借自身的推理能力，结合相关技术文档与经验知识，对答案进行补充完善，为用户提供全面且准确的解答。
以上功能，实现的关键原理是我预先定义了三个tools（排障、查询、定位），由LLM根据用户输入，自行决定选择工具并传递参数，最后由智能体执行函数。要实现这个流程，可以使用LLM自带的函数调用功能，也可以使用LangChain Tools（事实上我写好了的这个智能体，用的就是langchain）。但是3月12日，openai开源了OpenAI Agents SDK。我看了些up主的演示之后，自己也试了一下，感觉比之前两种方法都要更加简洁。于是决定一边写这篇文章，一边用penAI Agents SDK来重新编写我的这个AI智能体的代码。
OpenAI Agents SDK 是一个用于构建多代理工作流的轻量级但功能强大的框架,它的详细介绍可以参考官方文档Attention Required! | Cloudflare
下面我就直接进入实验和代码环节了。
首先，安装openai agents sdk
pip install openai-agents下面是参考官网的hello world代码：
from agents import Agent, Runner, RunConfig, OpenAIProvider from openai import AsyncOpenAI import asyncio from dotenv import load_dotenv from agents import set_default_openai_client from agents import set_default_openai_api from agents import set_tracing_disabled # 配置自定义 OpenAI 客户端 client = AsyncOpenAI(api_key="your_key",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1") # 设置默认的 OpenAI Agent 智能体客户端 set_default_openai_client(client) # 禁用跟踪 set_tracing_disabled(True) # 设置默认的 OpenAI API set_default_openai_api('chat_completions') # 创建简单的Agent agent = Agent( name="Assistant", instructions="You are a helpful assistant", model="qwen-turbo" ) # 使用同步方式运行 def main(): result = Runner.run_sync(agent, "你好，你是谁？") print(result.final_output) if __name__ == "__main__": main()输出内容如下：
在这里要说明的是，我目前使用的模型是阿里通义千问qwen-turbo，理论上你可以使用任何支持tools的LLM。而我在内网使用的是qwq-32B,同样能达到想要的效果。
如果我们想要像平时在web界面跟llm聊天的那种逐字输出的感觉，我们可以使用Runner.run_streamed创建一个RunResultstreaming对象，通过response.stream_events() 获取异步事件流，并使用 async for 遍历。
# 使用异步方式运行并处理流式输出 async def main(): # 创建 RunResultStreaming 对象 response = Runner.run_streamed(agent, "你好，你是谁？") # 异步遍历流式事件 async for event in response.stream_events(): # 检查事件类型并处理 if event.type == "raw_response_event" and hasattr(event.data, 'delta'): print(event.data.delta, end="", flush=True) Tools的存在使得agent可以调用外部功能（例如访问api、运行代码、执行搜索、查询数据库），通过将python函数公开为工具，LLM在运行时决定使用工具并传递需要调用的参数。
以下是官方给出的示范：
基于这个流程和格式，我编写了三个函数，具体内容没写，只是让他们在被调用时打印出相应内容，主要用于验证 LLM 能否依据用户输入准确调用函数并传递参数。
#查询某地点可用IP的工具 @function_tool def get_available_ip(location: str) -> str: """ 查询指定地点的可用ip """ print(f"我已提取地点:{location},并运行get_available_ip函数。") #根据IP查找定位终端的工具 @function_tool def iptracer(ip: str) -> str: """ 根据IP地址查找定位终端所在位置 """ print(f"我已提取IP:{ip},并运行iptracer函数。") # 排查指定地点网络故障的工具 @function_tool def troubleshooter(location: str) -> str: """ 排查指定地点的网络故障 """ print(f"我已提取地点:{location},并运行troubleshooter函数。") agent = Agent( name="Hello world", instructions="You are a helpful agent.", tools=[get_available_ip, iptracer, troubleshooter], model="qwen-turbo" ) async def main(): # 测试不同的输入 response = Runner.run_streamed(agent, "洛杉矶大厦网络中断了，请帮我排查") async for event in response.stream_events(): if event.type == "raw_response_event" and hasattr(event.data, 'delta'): print(event.data.delta, end="", flush=True) print("\n\n") response2 = Runner.run_streamed(agent, "查询上海广场可用IP") async for event in response2.stream_events(): if event.type == "raw_response_event" and hasattr(event.data, 'delta'): print(event.data.delta, end="", flush=True) print("\n\n") response3 = Runner.run_streamed(agent, "查找定位192.168.1.1") async for event in response3.stream_events(): if event.type == "raw_response_event" and hasattr(event.data, 'delta'): print(event.data.delta, end="", flush=True) print("\n\n") response4 = Runner.run_streamed(agent, "天空为什么是蓝色的？") async for event in response4.stream_events(): if event.type == "raw_response_event" and hasattr(event.data, 'delta'): print(event.data.delta, end="", flush=True) print("\n\n") if __name__ == "__main__": asyncio.run(main())在这段代码中，我定义了三个工具函数，并使用了 docstring。在 OpenAI Agents SDK 中，docstring 会被agent用来理解工具的作用。当agent接收到输入时（如“查询上海广场可用IP”），它会根据函数名和 docstring 判断是否调用此工具。代码工作的总体流程和关系如下：
- 用户输入 → agent根据 instructions 和工具的 docstring 决定行动。
- 工具调用 → 如果匹配到工具（如 troubleshooter），执行并输出结果。
- 流式输出 → 通过 Runner.run_streamed 和 stream_events() 实时显示。
- 无工具匹配 → 对于“天空为什么是蓝色的”，代理直接生成回答。
输出内容如下：
可以看到，LLM的输出是符合我们预期的，工具调用准确，打印的日志也正确。在最后一个问题，LLM没有匹配到工具时，agent直接生成自然语言回答，解释填空为什么是蓝色的。
今天先写这么多，具体的函数会在后面的文章中逐一细述，谢谢大家。
发布者：Ai探索者，转载请注明出处：https://javaforall.net/239133.html原文链接：https://javaforall.net
