Dify的Chatflow实现

在上一次的基础上进行一些功能实现。

通过向大模型描述请求来完成一些服务器端的功能

设计

流程 Dify Chatflow 任务流 → 自定义插件构建客户端请求 → 服务器端监听请求，解析请求并处理 →插件接收请求响应，传递结果给LLM 进行优化展示。

创建任务流

创建空白应用→Chatflow

ChatFlow是由多个组件构成的工作流程

在这个工作流程中，每次的对话会被作为输入参数传入到工作流中进行处理

当前实现的功能 ：

1. 查询服务器允许执行的命令行工具
2. 由大模型分析用户指令，从提供的命令列表中组织命令流程下发到服务器端执行

这里的编排

开始-问题分类器-情况1(向服务器端请求查询功能)-交由插件向服务器端发送查询请求-处理响应-回复
              -情况2(分析用户需求生成对应指令)-由插件构建请求向服务器请求执行-处理响应-回复

自定义插件向服务端发送请求

由 上一节中的设计流程来看我们至少需要构建两个请求 ：

1 . Get 请求 向服务端请求应用程序列表

#Desktop_get
import requests
import json

FetchKey = "/key"
Command = "/command"
class DestopHTool(Tool):
    def _invoke(self, tool_parameters: dict[str, Any]) -> Generator[ToolInvokeMessage]:
        ip = tool_parameters.get("ip","")
        port = tool_parameters.get("port","")
        key = self.fetch_key(ip,port)
        url = f"http://{ip}:{port}{Command}"
        headers={"X-Api-Key":key}
        res = requests.get(url=url,headers=headers,timeout=30).json()
        yield self.create_text_message(str(res["commands"])
        )

    def fetch_key(self,ip,port) : 
        url = f"http://{ip}:{port}{FetchKey}"

        js = requests.get(url=url,timeout=30).json()
        return js["key"]

2. Post 请求 向服务端发送命令执行列表，由服务端进行命令执行

import requests
import json

EXEC="/execute"

class DestopHelperTool(Tool):
    def _invoke(self, tool_parameters: dict[str, Any]) -> Generator[ToolInvokeMessage]:
        content = tool_parameters.get("text","")
        ip = tool_parameters.get("ip","")
        port = tool_parameters.get("port","")
        url = f"http://{ip}:{port}{EXEC}"
        print(url)
        key = tool_parameters.get("key","")
        headers={"X-Api-Key":key,"Content-Type":"application/json"}
        print(headers)
        res = requests.post(url=url,json=json.loads(content),headers=headers,timeout=30).text
        yield self.create_text_message(res)

构建服务器端接口

使用flask 构建一个接收以上请求的服务程序。

from flask import Flask, jsonify, request
from functools import wraps
import uuid
import subprocess
import pexpect

app = Flask(__name__)

def require_key(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        print(request.headers)
        key = request.headers.get('X-API-KEY')
        print(keys)
        if key not in keys:
            return jsonify({'error': 'Invalid API key'}), 401
        return f(*args, **kwargs)
    return decorated_function

# 密钥创建路由
@app.route('/key', methods=['GET'])
def create_key():
    key = str(uuid.uuid4())
    keys[key] = True
    print(keys)
    return jsonify({'key': key})

# 获取所有功能路由
@app.route('/command', methods=['GET'])
@require_key
def get_functions():
    try:
        # 获取PATH环境变量中的所有可执行文件
        result = subprocess.run(['ls', '/server/bin'], capture_output=True, text=True)
        commands = result.stdout.split('\n')
        return jsonify({'status': 'success', 'commands': commands})
    except Exception as e:
        return jsonify({'status': 'error', 'message': str(e)})
# 执行命令路由
@app.route('/execute', methods=['POST'])
@require_key
def execute():
    """
    执行传入的命令并返回结果
    """
    data = request.json
    print(data)
    commands = data.get('commands')
    result = []
    if not commands:
        return jsonify({'status': 'error', 'message': 'No command provided'})
    
    try:
        shell = pexpect.spawn('/bin/bash')
        for command in commands:
            if not isinstance(command, dict):
                return jsonify({'status': 'error','message': 'Invalid command format'})
            command_line = []
            command_line.append(command["command"])
            command_line.extend(command["args"])
            cmd = " ".join(command_line)
            marker = f"CMD_END_{hash(cmd)}"
            shell.sendline(f"{cmd}; echo '{marker}'")
            shell.expect(marker)
            # 获取输出
            output = shell.before.decode().strip()
            print(output)
            result.append( {
                'command': cmd,
                'status':'success',
                'stdout': output,
            })
    except Exception as e:
        return jsonify({'status': 'error', 'message': str(e)})
    shell.close()
    print(result)
    return jsonify({"result": result})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

Get 请求返回的消息体

{
    "commands": [
        "hexo",
        "ls",
        "lscpu",
        "python",
        "zip",
        ""
    ],
    "status": "success"
}

Post请求接收的消息体

{
  "commands": [
    {
      "command": "ls",
      "args": ["-l", "/home/codfish"]
    }
  ]
}

启动本地插件调试和服务端

执行工作流

向机器人请求查看服务端提供的应用列表 ：

向机器人请求查询某个目录下的文件信息：

请求安装python 库：

总结

完成了基本功能的实现

在某些情况下提示词还是会出现异常，需要进一步优化。