利用dify打造命令行助手

利用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

1. 将结果喂给ai客户端

   这就没有任何花稍的技术可言了，打deepseek,或豆包。描述背景：我执行了XXXX命令，得到结果如下：（然后cmd + c）请帮我统计每个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()
   ​

1. 利用别名将脚本与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命令

结果演示