利用dify打造命令行助手

前言

我是mac os 的使用者。 如果说macos哪个工具我使用最频繁，最能提高效率的工作工具非zsh莫属(当然，我安装了iterm)。前不久，我要扩容线上的k8s集群。便想统计一下每个node上运行的pod数量。我不知道合适的命令，但我知道获取所有pod的命令，我想的法是用kuectl 获取所有的pod列表， 然后将结果拷贝到某个ai工具中，让他帮我统计。

步骤如下：

1. 执行命令兼拷贝结果

kubectl get po -A -o wide |  pbcopy

后面我可以把命令变得更高级点

#在.zsh文件中创建这个函数
function pc(){
	tee >(pbcopy)
}

便可以执行这个命令，即能得到标准输出，又能将输出结果拷贝到剪切板。

kubectl get po -A -o wide |  pc

2. 将结果喂给ai客户端

   这就没有任何花稍的技术可言了，打deepseek,或豆包。描述背景：我执行了XXXX命令，得到结果如下：（然后cmd + v）请帮我统计每个node上运行的pod数量，用表格显示

如上方法太浪费时间了，且我要暂时离开我亲爱的命令行工具，还要别启软件，说一堆背说描述。试想如果我能直接用命令行与ai交流岂不美哉。正好我在看dify之类的工具。便想随便做一个玩玩。

正文

1. 利用dify定义一个agent

创建一个聊天助手型的空白应用。输入提示词如下

你是一个资深运维，根据用户运行的命令:{{command}}及运行的结果:{{result}}，回答用户的问题

我选择模型提供者是 groq cloud。（因为快）。模型随便选了一个.

2. 捕捉用户的命令及运行结果并记录

在调用上面的ai应用前，我要获取两个输入参数，当前执行的命令，及命令结果

1. 获取当前的命令

我记得zsh提供了一些勾子函数，查了一下，果然后。利用勾子函数，在每次命令执行后，将命令记录到文本文件中

preexec() {
    # 记录命令到日志文件
    echo $1 >> command_log.txt
}

后文我会给出完整代码

2. 获取输出结果

和我在前言中的使用的方法一样，可以用管道获取

命令 | tee result.txt

后文我会给出完整代码

3. 用python获取参数，并调用dify的接口

assistant-chat.py 文件：

import sys
import json
import requests
import threading
token = 'app-2GuvOUR6EO5lUo2pm7EjuUwv'


class AssistantChat:
    def __init__(self, command, result):
        self.command = command
        self.result = result
        self.question = None
        self.conversation_id = None

        self.headers = {"Content-Type": "application/json",
                       "Authorization": f"Bearer {token}"}
        self.url = 'http://localhost/v1/chat-messages'
    
    def ask(self, question):
        inputs_data = {'command':self.command,'result':self.result}
        request_data = {
            'inputs':inputs_data,
            'user': 'user-123',
            'conversation_id': self.conversation_id,
            'query':question
        }  

    
       
        self.send(request_data)

       # print(json.dumps(request_data)) 

    def send(self, request_data):
        response = requests.post(self.url, json=request_data,headers=self.headers)
        if response.status_code == 200:
            response_data = response.json()
            # print(response_data)
            answer = response_data['answer']
            print(answer)
           
            self.conversation_id = response_data['conversation_id']
            return response_data
        else:
            print(f"Error: {response.content}")
            print(f"Error: {response.status_code}")   

    def run(self):
        while True:
            try:
                question = input("请输入您的问题: ")
                self.ask(question)
            except KeyboardInterrupt:
                break


def main():
    print("=============================")
    ## 读取 result.txt 文件内容
    with open('result.txt', 'r') as file:
        result = file.read()
        #print(result)
    
    with open('command_log.txt') as file:
        commands = file.readlines()
        cmd = commands[len(commands)-1]
        #cmd ="kubectl get po -A -o wide | mychat"
        #截取cmd到|
        cmd = cmd[:cmd.find('|')]
        #print(cmd)

    assistant = AssistantChat(cmd, result)
    assistant.run()


if __name__ == '__main__':
    main()

4. 利用别名将脚本与python代码串联系起来

alias mychat='tee result.txt && python assistant-chat.py'

完整的shell脚本如下：

assistant-chat.zsh

preexec() {
    # 记录命令到日志文件
    echo $1 >> command_log.txt
}

alias mychat='tee result.txt && python assistant-chat.py'

在使用之前，需要执行source命令

结果演示