本次LangChain搭建故障诊断Agent,用Trea和Cursor自动生成代码:

帮我搭建一个网络故障诊断类的Agent,工具之间存在串联,第一个工具的输出是第二个工具的输入;如果有10个工具的话,会有很多输入输出的关系。希望Agent,在适当的时候调用适合的工具;===可以参考@*******_toolchain.py 编写新的Python

生成的完整代码:

import osimport jsonimport refrom typing importList, Union, Dict, Any, Optionalfrom langchain.agents import Tool, AgentExecutor, create_react_agent, AgentType, initialize_agentfrom langchain.prompts import PromptTemplatefrom langchain.schema import AgentAction, AgentFinishfrom langchain_community.llms import Tongyifrom langchain.memory import ConversationBufferMemory# --- 环境设置 ---# 确保设置了您的通义千问 API 密钥# 您可以通过环境变量 DASHSCOPE_API_KEY 设置,或者直接在这里修改# DASHSCOPE_API_KEY = os.getenv('DASHSCOPE_API_KEY', '您的API密钥')DASHSCOPE_API_KEY = os.getenv('DASHSCOPE_API_KEY') # 请替换为您的有效密钥# --- 自定义网络诊断工具 ---classPingTool:    """执行 Ping 操作以检查与目标的网络连通性。"""    def__init__(self):        self.name = "网络连通性检查 (Ping)"        self.description = "检查本机到指定主机名或 IP 地址的网络连通性。输入应该是目标主机名或 IP 地址。输出表明是否可达及延迟。"    defrun(self, target: str) -> str:        """模拟执行 ping 命令。        参数:            target: 目标主机名或 IP 地址。        返回:            模拟的 ping 结果。        """        print(f"--- 模拟执行 Ping: {target} ---")        # 简单模拟:特定目标可能失败        if"unreachable"in target or target == "192.168.1.254":            returnf"Ping {target} 失败:请求超时。"        elif target == "localhost"or target == "127.0.0.1":             returnf"Ping {target} 成功:延迟 <1ms。"        elif"example.com"in target:             # 模拟一个稍微慢点的响应             import random             delay = random.randint(20, 80)             returnf"Ping {target} 成功:延迟 {delay}ms。"        else:            # 其他情况默认成功            import random            delay = random.randint(5, 50)            returnf"Ping {target} 成功:延迟 {delay}ms。"classDNSTool:    """执行 DNS 查询以解析主机名。"""    def__init__(self):        self.name = "DNS解析查询"        self.description = "解析给定的主机名,获取其对应的 IP 地址。输入应该是要解析的主机名。输出是 IP 地址或解析失败信息。"    defrun(self, hostname: str) -> str:        """模拟 DNS 查询。        参数:            hostname: 要解析的主机名。        返回:            模拟的 DNS 解析结果。        """        print(f"--- 模拟 DNS 查询: {hostname} ---")        # 简单模拟        if hostname == "www.example.com":            returnf"DNS 解析 {hostname} 成功:IP 地址是 93.184.216.34"        elif hostname == "internal.service.local":            returnf"DNS 解析 {hostname} 成功:IP 地址是 192.168.1.100"        elif hostname == "unknown.domain.xyz":            returnf"DNS 解析 {hostname} 失败:找不到主机。"        elif hostname == "127.0.0.1"or re.match(r"\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}", hostname):             returnf"输入 '{hostname}' 已经是 IP 地址,无需 DNS 解析。"        else:            # 模拟一个通用 IP            returnf"DNS 解析 {hostname} 成功:IP 地址是 10.0.0.5"classInterfaceCheckTool:    """检查本地网络接口的状态。"""    def__init__(self):        self.name = "本地网络接口检查"        self.description = "检查本机网络接口的状态(如 IP 地址、是否启用)。可选输入是接口名称,若不提供则检查默认接口。输出接口状态信息。"    defrun(self, interface_name: Optional[str] = None) -> str:        """模拟检查网络接口状态。        参数:            interface_name: (可选) 要检查的接口名称。        返回:            模拟的接口状态信息。        """        print(f"--- 模拟检查接口状态: {interface_name or'默认接口'} ---")        if interface_name and"eth1"in interface_name.lower():            returnf"接口 '{interface_name}' 状态:关闭 (Administratively down)"        else:            # 模拟一个常见的有线或无线接口状态            returnf"接口 'Ethernet'/'Wi-Fi' 状态:启用, IP 地址: 192.168.1.50, 子网掩码: 255.255.255.0, 网关: 192.168.1.1"classLogAnalysisTool:    """分析系统或应用日志以查找网络相关的错误。"""    def__init__(self):        self.name = "网络日志分析"        self.description = (            "搜索系统或应用程序日志,查找与网络问题相关的条目。"            "输入应该是描述问题的关键词(例如 'timeout', 'connection refused', 'dns error')和可选的时间范围。"            "输出是找到的相关日志条目摘要或未找到相关条目的消息。"        )    defrun(self, keywords: str, time_range: Optional[str] = "过去1小时") -> str:        """模拟分析日志。        参数:            keywords: 用于搜索日志的关键词。            time_range: (可选) 要搜索的时间范围描述,默认为 '过去1小时'。        返回:            模拟的日志分析结果。        """        print(f"--- 模拟分析日志: 关键词='{keywords}', 时间范围='{time_range}' ---")        # 简单模拟        keywords_lower = keywords.lower()        if"timeout"in keywords_lower or"超时"in keywords_lower:            return (f"在 {time_range} 的日志中找到 3 条与 '{keywords}' 相关的条目:\n"                    f"- [Error] 连接到 10.0.0.88:8080 超时\n"                    f"- [Warning] 对 api.external.com 的请求超时\n"                    f"- [Error] 内部服务通信超时")        elif"connection refused"in keywords_lower or"连接被拒绝"in keywords_lower:             return (f"在 {time_range} 的日志中找到 1 条与 '{keywords}' 相关的条目:\n"                     f"- [Error] 连接到 192.168.1.200:5432 失败:Connection refused")        elif"dns"in keywords_lower:             return (f"在 {time_range} 的日志中找到 2 条与 '{keywords}' 相关的条目:\n"                     f"- [Warning] DNS 服务器 8.8.8.8 响应慢\n"                     f"- [Error] 无法解析主机名 'failed.internal.service'")        else:            returnf"在 {time_range} 的日志中未找到与 '{keywords}' 相关的明显网络错误条目。"# --- 创建 Agent 和工具链 ---defcreate_network_diagnosis_chain():    """创建网络故障诊断的 Agent 执行器。"""    # 1. 初始化工具    ping_tool = PingTool()    dns_tool = DNSTool()    interface_tool = InterfaceCheckTool()    log_tool = LogAnalysisTool()    # 2. 将工具类包装成 LangChain Tool 对象    # 注意:这里的 description 对 Agent 如何使用工具至关重要    tools = [        Tool(            name=ping_tool.name,            func=ping_tool.run,            description=ping_tool.description        ),        Tool(            name=dns_tool.name,            func=dns_tool.run,            description=dns_tool.description        ),        Tool(            name=interface_tool.name,            func=interface_tool.run,            description=interface_tool.description        ),        Tool(            name=log_tool.name,            func=log_tool.run,            description=log_tool.description        )        # 如果有更多工具(如 Traceroute, Get Config等),在这里添加    ]    tool_names = ", ".join([t.name for t in tools])    # 3. 初始化语言模型 (使用通义千问)    llm = Tongyi(model_name="qwen-turbo", dashscope_api_key=DASHSCOPE_API_KEY)    # 4. 创建 Zero-Shot ReAct Agent    agent = initialize_agent(        tools,        llm,        agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,        verbose=True,        handle_parsing_errors=False,        memory=ConversationBufferMemory(memory_key="chat_history", input_key="input", output_key="output"),        max_iterations=10    )    return agent# --- 示例:使用 Agent 处理网络诊断任务 ---defdiagnose_network_issue(issue_description: str):    """    使用网络诊断 Agent 处理用户报告的网络问题。    参数:        issue_description: 用户描述的网络问题。    返回:        Agent 的最终诊断结果。    """    try:        print(f"\\n--- 开始诊断任务 ---")        print(f"用户问题: {issue_description}")        agent = create_network_diagnosis_chain()        # 使用 invoke 而不是 run,以便更好地处理输入输出和记忆        response = agent.invoke({"input": issue_description})        # response 是一个字典,我们通常关心 "output" 键        return response.get("output", "未能获取诊断结果。")    except Exception as e:        # 打印更详细的错误信息,有助于调试        import traceback        print(f"处理诊断任务时发生错误: {str(e)}")        traceback.print_exc()        returnf"处理诊断任务时出错: {str(e)}"# --- 主程序入口 ---if __name__ == "__main__":    # 示例 1: 无法访问特定网站    task1 = "我无法访问 www.example.com,浏览器显示连接超时。"    print("诊断任务 1:")    result1 = diagnose_network_issue(task1)    print("\\n--- 诊断任务 1 结束 ---")    print(f"最终诊断结果: {result1}")    print("\\n" + "="*50 + "\\n") # 分隔符    # 示例 2: 内部服务访问失败    task2 = "连接到内部数据库服务器 (internal.service.local) 失败,提示 'connection refused'。"    print("诊断任务 2:")    result2 = diagnose_network_issue(task2)    print("\\n--- 诊断任务 2 结束 ---")    print(f"最终诊断结果: {result2}")    # # 示例 3: DNS 解析问题 (需要 DNSTool 模拟失败)    # task3 = "我打不开网站 unknown.domain.xyz,好像是 DNS 问题。"    # print("诊断任务 3:")    # result3 = diagnose_network_issue(task3)    # print("\\n--- 诊断任务 3 结束 ---")    # print(f"最终诊断结果: {result3}")    # # 示例 4: 本地网络接口问题 (需要 InterfaceCheckTool 模拟失败)    # task4 = "我的电脑连不上网了,检查一下接口 eth1 的状态。"    # print("诊断任务 4:")    # result4 = diagnose_network_issue(task4)    # print("\\n--- 诊断任务 4 结束 ---")    # print(f"最终诊断结果: {result4}") 

代码中没有用到create_react_agent,create_react_agent:需要在提示词中,指明 {tools} 和 {tool_names},代码中用到的是ZERO_SHOT_REACT_DESCRIPTION是自己去对工具做判断,不需要自己再把工具名字写一遍。代码中的ConversationBufferMemory(memory_key="chat_history", input_key="input", output_key="output")把上下文记忆带进去max_iterations=10最多十次迭代

诊断工具详解:

(1)网络连通性检查工具 (PingTool)

功能:检查从本机到指定主机名或 IP 的网络连通性

输入:目标主机名或 IP 地址

输出:连通性状态(成功/失败)和网络延迟

使用场景:验证网络连接是否通畅,检测网络延迟

(2)DNS 解析查询工具 (DNSTool)

功能:将主机名解析为 IP 地址

输入:需要解析的主机名

输出:解析后的 IP 地址或解析失败信息

使用场景:诊断 DNS 解析问题,为后续的连通性测试提供 IP 地址

(3)本地网络接口检查工具 (InterfaceCheckTool)

功能:检查本地网络接口的状态

输入:(可选)接口名称

输出:接口状态信息(是否启用、IP 地址、子网掩码等)

使用场景:检查本地网络配置是否正确

(4)网络日志分析工具 (LogAnalysisTool)

功能:在系统或应用日志中搜索网络相关问题

输入:关键词和可选的时间范围

输出:匹配的日志条目

使用场景:查找历史网络错误记录,发现非实时故障的线索

图片

代码执行过程和结果(工具的调用顺序是大模型自己判断的):

诊断任务 1:\n--- 开始诊断任务 ---用户问题: 我无法访问 www.example.com,浏览器显示连接超时。d:\learn\langchain\CASE-搭建故障诊断Agent\2-network_diagnosis_agent.py:185: LangChainDeprecationWarning: Please see the migration guide at: https://python.langchain.com/docs/versions/migrating_memory/  memory=ConversationBufferMemory(memory_key="chat_history", input_key="input", output_key="output"),d:\learn\langchain\CASE-搭建故障诊断Agent\2-network_diagnosis_agent.py:179: LangChainDeprecationWarning: LangChain agents will continue to be supported, but it is recommended fornew use cases to be built with LangGraph. LangGraph offers a more flexible and full-featured framework for building agents, including support for tool-calling, persistence of state, and human-in-the-loop workflows. For details, refer to the `LangGraph documentation <https://langchain-ai.github.io/langgraph/>`_ as well as guides for `Migrating from AgentExecutor <https://python.langchain.com/docs/how_to/migrate_agent/>`_ and LangGraph's `Pre-built ReAct agent <https://langchain-ai.github.io/langgraph/how-tos/create-react-agent/>`_.  agent = initialize_agent(

> Entering new AgentExecutor chain...我需要检查本机到 www.example.com 的网络连通性。Action: 网络连通性检查 (Ping)Action Input: www.example.com--- 模拟执行 Ping: www.example.com ---
Observation: Ping www.example.com 成功:延迟 60ms。Thought:网络连通性检查显示可以成功访问 www.example.com,因此问题可能不在于网络连接。接下来我需要确认 www.example.com 的 IP 地址以确保 DNS 解析没有问题。Action: DNS解析查询Action Input: www.example.com--- 模拟 DNS 查询: www.example.com ---
Observation: DNS 解析 www.example.com 成功:IP 地址是 93.184.216.34Thought:DNS 解析成功,表明域名解析没有问题。由于网络连通性和 DNS 解析均正常,问题可能出在浏览器或服务器端配置上。Final Answer: 您的网络连通性和 DNS 解析均正常,问题可能与浏览器设置、服务器端配置或网站暂时不可用有关。建议尝试清除浏览器缓存、更换浏览器或稍后再试。如果问 题仍然存在,请联系 www.example.com 的技术支持团队以确认服务器端是否有问题。
> Finished chain.\n--- 诊断任务 1 结束 ---最终诊断结果: 您的网络连通性和 DNS 解析均正常,问题可能与浏览器设置、服务器端配置或网站暂时不可用有关。建议尝试清除浏览器缓存、更换浏览器或稍后再试。如果问 题仍然存在,请联系 www.example.com 的技术支持团队以确认服务器端是否有问题。\n==================================================\n诊断任务 2:\n--- 开始诊断任务 ---用户问题: 连接到内部数据库服务器 (internal.service.local) 失败,提示 'connection refused'。

> Entering new AgentExecutor chain... 需要先检查到目标服务器的网络连通性,以确认是否可以到达该服务器。Action: 网络连通性检查 (Ping)Action Input: internal.service.local--- 模拟执行 Ping: internal.service.local ---
Observation: Ping internal.service.local 成功:延迟 19ms。Thought:网络连通性没有问题,接下来需要确认目标主机名解析的 IP 地址是否正确。Action: DNS解析查询Action Input: internal.service.local--- 模拟 DNS 查询: internal.service.local ---
Observation: DNS 解析 internal.service.local 成功:IP 地址是 192.168.1.100Thought:DNS 解析也没有问题,目标主机名解析到的 IP 地址是 192.168.1.100。接下来需要检查本地网络接口的状态,确保本机配置正确。Action: 本地网络接口检查Action Input: None--- 模拟检查接口状态: None ---
Observation: 接口 'Ethernet'/'Wi-Fi' 状态:启用, IP 地址: 192.168.1.50, 子网掩码: 255.255.255.0, 网关: 192.168.1.1Thought:本地网络接口状态正常,IP 地址为 192.168.1.50,与目标服务器在同一子网内。网络连通性和 DNS 解析均无问题,因此需要进一步检查日志以了解 'connection refused' 的原因。Action: 网络日志分析Action Input: keywords: 'connection refused', time_range: '过去1小时'--- 模拟分析日志: 关键词='keywords: 'connection refused', time_range: '过去1小时'', 时 间范围='过去1小时' ---
Observation: 在 过去1小时 的日志中找到 1 条与 'keywords: 'connection refused', time_range: '过去1小时'' 相关的条目:- [Error] 连接到 192.168.1.200:5432 失败:Connection refusedThought:日志显示连接到 192.168.1.200:5432 时出现 'Connection refused' 错误。这表明问题可能与目标服务器上的服务状态或端口配置有关,而不是网络连通性问题。    Final Answer: 网络连通性和 DNS 解析均正常,但目标服务器 (internal.service.local, IP 地址 192.168.1.100) 上的服务可能未运行或未监听正确的端口。建议检查数据库服务器的服务状态和防火墙配置,确保其正在监听正确的端口并且允许来自客户端的连接。
> Finished chain.\n--- 诊断任务 2 结束 ---最终诊断结果: 网络连通性和 DNS 解析均正常,但目标服务器 (internal.service.local, IP 地址 192.168.1.100) 上的服务可能未运行或未监听正确的端口。建议检查数据库服务器的服务状态和防火墙配置,确保其正在监听正确的端口并且允许来自客户端的连接。

工具链接示例

系统能够实现工具间的串联调用,例如:

```

问题:无法访问网站 www.example.com

DNS解析查询(www.example.com) → 返回 IP: 93.184.216.34

网络连通性检查(93.184.216.34) → 返回连接超时

本地网络接口检查() → 返回接口正常

网络日志分析("timeout") → 发现相关错误日志

诊断结论:可能是网络路由问题或目标服务器不可用

```

这个例子中,可以使用 AgentType.ZERO_SHOT_REACT_DESCRIPTION这是 LangChain 框架内置的一种 Agent 类型,属于“Zero-Shot ReAct”范式。Zero-Shot指的是大模型在没有额外训练或示例的情况下,直接根据提示词(Prompt)和工具描述来推理如何调用工具。

ReAct:是一种“思考-行动-观察”循环范式(Reasoning + Acting),即

• 模型会先思考(Thought),

• 再决定调用哪个工具(Action),

• 然后观察工具的输出(Observation),

• 再继续推理,直到得出结论。

AgentType.ZERO_SHOT_REACT_DESCRIPTION 的特点:

自动读取你传入的工具(Tool)列表,每个工具的name 和 description 会被自动拼接到系统提示词中。

Agent 会根据用户输入和工具描述,自动决定调用哪个工具、如何调用。

不需要自己写复杂的 Prompt,LangChain 内部会自动生成一份标准的英文提示词,格式大致如下:

Answer the following questions as best you can. You have access to the following tools:<工具1名称>: <工具1描述><工具2名称>: <工具2描述>...Use the following format:Question: the input question you must answerThought: you should always think about what to doAction: the action to take, should be one of [工具1, 工具2, ...]Action Input: the input to the actionObservation: the result of the action... (this Thought/Action/Action Input/Observation can repeat N times)Thought: I now know the final answerFinal Answer: the final answer to the original input question

 一、大模型风口已至:月薪30K+的AI岗正在批量诞生

2025年大模型应用呈现爆发式增长,根据工信部最新数据:

国内大模型相关岗位缺口达47万

初级工程师平均薪资28K

70%企业存在"能用模型不会调优"的痛点

真实案例:某二本机械专业学员,通过4个月系统学习,成功拿到某AI医疗公司大模型优化岗offer,薪资直接翻3倍!

二、如何学习大模型 AI ?


🔥AI取代的不是人类,而是不会用AI的人!麦肯锡最新报告显示:掌握AI工具的从业者生产效率提升47%,薪资溢价达34%!🚀

由于新岗位的生产效率,要优于被取代岗位的生产效率,所以实际上整个社会的生产效率是提升的。

但是具体到个人,只能说是:

“最先掌握AI的人,将会比较晚掌握AI的人有竞争优势”。

这句话,放在计算机、互联网、移动互联网的开局时期,都是一样的道理。

我在一线互联网企业工作十余年里,指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家,也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑,所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限,很多互联网行业朋友无法获得正确的资料得到学习提升,故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

1️⃣ 提示词工程:把ChatGPT从玩具变成生产工具
2️⃣ RAG系统:让大模型精准输出行业知识
3️⃣ 智能体开发:用AutoGPT打造24小时数字员工

📦熬了三个大夜整理的《AI进化工具包》送你:
✔️ 大厂内部LLM落地手册(含58个真实案例)
✔️ 提示词设计模板库(覆盖12大应用场景)
✔️ 私藏学习路径图(0基础到项目实战仅需90天)

 

第一阶段(10天):初阶应用

该阶段让大家对大模型 AI有一个最前沿的认识,对大模型 AI 的理解超过 95% 的人,可以在相关讨论时发表高级、不跟风、又接地气的见解,别人只会和 AI 聊天,而你能调教 AI,并能用代码将大模型和业务衔接。

*   大模型 AI 能干什么?
*   大模型是怎样获得「智能」的?
*   用好 AI 的核心心法
*   大模型应用业务架构
*   大模型应用技术架构
*   代码示例:向 GPT-3.5 灌入新知识
*   提示工程的意义和核心思想
*   Prompt 典型构成
*   指令调优方法论
*   思维链和思维树
*   Prompt 攻击和防范
*   …

第二阶段(30天):高阶应用

该阶段我们正式进入大模型 AI 进阶实战学习,学会构造私有知识库,扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架,抓住最新的技术进展,适合 Python 和 JavaScript 程序员。

*   为什么要做 RAG
*   搭建一个简单的 ChatPDF
*   检索的基础概念
*   什么是向量表示(Embeddings)
*   向量数据库与向量检索
*   基于向量检索的 RAG
*   搭建 RAG 系统的扩展知识
*   混合检索与 RAG-Fusion 简介
*   向量模型本地部署
*   …

第三阶段(30天):模型训练

恭喜你,如果学到这里,你基本可以找到一份大模型 AI相关的工作,自己也能训练 GPT 了!通过微调,训练自己的垂直大模型,能独立训练开源多模态大模型,掌握更多技术方案。

到此为止,大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗?

*   为什么要做 RAG
*   什么是模型
*   什么是模型训练
*   求解器 & 损失函数简介
*   小实验2:手写一个简单的神经网络并训练它
*   什么是训练/预训练/微调/轻量化微调
*   Transformer结构简介
*   轻量化微调
*   实验数据集的构建
*   …

第四阶段(20天):商业闭环

对全球大模型从性能、吞吐量、成本等方面有一定的认知,可以在云端和本地等多种环境下部署大模型,找到适合自己的项目/创业方向,做一名被 AI 武装的产品经理。

*   硬件选型
*   带你了解全球大模型
*   使用国产大模型服务
*   搭建 OpenAI 代理
*   热身:基于阿里云 PAI 部署 Stable Diffusion
*   在本地计算机运行大模型
*   大模型的私有化部署
*   基于 vLLM 部署大模型
*   案例:如何优雅地在阿里云私有部署开源大模型
*   部署一套开源 LLM 项目
*   内容安全
*   互联网信息服务算法备案
*   …

学习是一个过程,只要学习就会有挑战。天道酬勤,你越努力,就会成为越优秀的自己。

如果你能在15天内完成所有的任务,那你堪称天才。然而,如果你能完成 60-70% 的内容,你就已经开始具备成为一名大模型 AI 的正确特征了。

这份完整版的大模型 AI 学习资料已经上传CSDN,朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

Logo

火山引擎开发者社区是火山引擎打造的AI技术生态平台,聚焦Agent与大模型开发,提供豆包系列模型(图像/视频/视觉)、智能分析与会话工具,并配套评测集、动手实验室及行业案例库。社区通过技术沙龙、挑战赛等活动促进开发者成长,新用户可领50万Tokens权益,助力构建智能应用。

更多推荐