本地準備

要求：

接口開發

讓kimi給我生成一個文章表，并且插入10條數據，我們可以告訴kimi，文章長度多大，這樣內容可以豐富些。

直接讓kimi生成一個暴露接口查數據庫的服務，有簡單的優化了下,將下面的內容放入到server.py文件中

from flask import Flask, request, jsonify

import pymysql



app = Flask(__name__)



# 數據庫配置

DATABASE_CONFIG = {

    'host': '', # 自己的數據庫地址

    'user': '', # 自己數據庫的賬戶

    'password': '', #自己數據庫的密碼

    'db': 'demo', # 自己數據庫的庫名

    'charset': 'utf8mb4',

    'cursorclass': pymysql.cursors.DictCursor

}



@app.route('/query', methods=['POST'])

def query_database():

    print("接收到請求")

    # 獲取關鍵字

    keyword = request.json.get('keyword')

    print("keyword為："+keyword)

    ifnot keyword:

        return jsonify({"error": "Keyword is required"}), 400



    # 參數化查詢，避免 SQL 注入，修改成自己的庫

    query = "SELECT * FROM articles WHERE content LIKE %s"

    params = ('%' + keyword + '%',)



    try:

        # 建立數據庫連接

        connection = pymysql.connect(**DATABASE_CONFIG)

        with connection.cursor() as cursor:

            # 執行查詢

            cursor.execute(query, params)

            result = cursor.fetchall()



        connection.commit()

        connection.close()



        ifnot result:

            return"未查詢到有效數據", 400



        # 生成 Markdown 表格

        markdown_table = generate_markdown_table(result)



        return markdown_table, 200



    except Exception as e:

        return str(e), 500



def generate_markdown_table(results):

    """ 生成 Markdown 表格 """

    ifnot results:

        return""

    # 獲取列名

    columns = results[0].keys()



    # 表頭

    table_md = "| " + " | ".join([col for col in columns]) + " |\n"

    # 分隔線

    table_md += "| " + " --- |" * len(columns) + "\n"



    # 表格內容

    for row in results:

        table_md += "| " + " | ".join([str(cell) for cell in row.values()]) + " |\n"



    return table_md



if __name__ == '__main__':

    # 注意這里綁定本機的內容ip，省事點，就0.0.0.0即可。不要綁定127.0.0.1，docke內訪問不到

    app.run(host='10.1.0.65', port=8000)

啟動服務

python .\server.py

配置工作流

創建一個空白應用。

在開始節點添加一個輸入字段context

添加一個LLM，把開始節點設置的context字段作為上下文傳入，并設置提示詞提取關鍵詞。

添加一個http請求節點，把我們在接口開發里的地址和接口名填寫進去2，然后把大模型的輸出作為關鍵詞填寫到請求body里3，我們關閉重試機制4。

這里要注意下：json的引號是中文的，最好在外面寫好校驗過了再放進去。

在HTTP請求的輸出變量里，我們只關注status_code 響應狀態碼和響應內容即可。

添加一個條件分支1，然后設置HTTP響應碼為200的時候，連接到大模型。其他直接結束。

添加大模型，將HTTP請求的響應體作為上下文給大模型，輸入提示詞，讓大模型根據知識，驗證，并進行合理性的驗證，最后結構化返回。

在結束節點中，我們把大模型整理的內容輸出。

試運行效果。

基于代碼執行查詢數據庫

由于difysandbox的安全限制

官方也有了對應的說明，見文檔。 https://github.com/langgenius/dify-sandbox/blob/main/FAQ.mdss

difysandbox源碼修改

一定要使用linux環境、一定要使用linux環境一定要使用linux環境

我從github上拉下代碼以后，搜索“syscalls_amd64.go

一共有4個文件，

我用python，不是arm架構的，鏡像都是linux的。

我們直接問kimi即可。

ps:這個問題丟給了ds和chatgpt都是瞎回答

一步步的問kimi，最后告訴我要添加哪些。整理以后添加到代碼里。

var ALLOW_SYSCALLS = []int{  

    // file io  

    syscall.SYS_NEWFSTATAT, syscall.SYS_IOCTL, syscall.SYS_LSEEK, syscall.SYS_GETDENTS64,  

    syscall.SYS_WRITE, syscall.SYS_CLOSE, syscall.SYS_OPENAT, syscall.SYS_READ,  

    // thread  

    syscall.SYS_FUTEX,  

    // memory  

    syscall.SYS_MMAP, syscall.SYS_BRK, syscall.SYS_MPROTECT, syscall.SYS_MUNMAP, syscall.SYS_RT_SIGRETURN,  

    syscall.SYS_MREMAP,  



    // user/group  

    syscall.SYS_SETUID, syscall.SYS_SETGID, syscall.SYS_GETUID,  

    // process  

    syscall.SYS_GETPID, syscall.SYS_GETPPID, syscall.SYS_GETTID,  

    syscall.SYS_EXIT, syscall.SYS_EXIT_GROUP,  

    syscall.SYS_TGKILL, syscall.SYS_RT_SIGACTION, syscall.SYS_IOCTL,  

    syscall.SYS_SCHED_YIELD,  

    syscall.SYS_SET_ROBUST_LIST, syscall.SYS_GET_ROBUST_LIST, SYS_RSEQ,  



    // time  

    syscall.SYS_CLOCK_GETTIME, syscall.SYS_GETTIMEOFDAY, syscall.SYS_NANOSLEEP,  

    syscall.SYS_EPOLL_CREATE1,  

    syscall.SYS_EPOLL_CTL, syscall.SYS_CLOCK_NANOSLEEP, syscall.SYS_PSELECT6,  

    syscall.SYS_TIME,  



    syscall.SYS_RT_SIGPROCMASK, syscall.SYS_SIGALTSTACK, SYS_GETRANDOM,  

    //新增  

    5, 6, 7, 21, 41, 42, 44, 45, 51, 54, 55, 107, 137, 204, 281,  

}

預裝mysql操作包

既然我們要操作在沙箱里操作mysql，那我們得在對應的環境中預裝下mysql客戶端。

在1對應的文件中添加2對應的pymysql==1.1.1，我直接安裝最新版。

在readme中有操作步驟

### Steps



1. Clone the repository using git clone https://github.com/langgenius/dify-sandbox and navigate to the project directory.
2. Run ./install.sh to install the necessary dependencies.
3. Run ./build/build_[amd64|arm64].sh to build the sandbox binary.
4. Run ./main to start the server.

編譯成功以后，打包鏡像。因為我沒有環境，直接模擬了下創建了一個main和env目錄

然后模擬打包鏡像。在根目錄中執行下面的命令

docker build -f docker/amd64/dockerfile -t dify-sandbox:local .

我在win上打包報了一堆錯，都扔給kimi，一步步的解決。最后成功。

沙箱網咯策略配置

在我們的安裝dify的的時候，有個dify/docker/ssrf_proxy目錄，找到squid.conf.template

在這里，你可以設置允許訪問的網絡，允許訪問的端口，生產一定要最小權限

acl devnet src 10.1.0.0/24

acl devnet src 10.255.200.0/24



acl Safe_ports port 3306        # MySQL

acl Safe_ports port 5432        # Postgres

acl Safe_ports port 27017       # MongoDB

acl Safe_ports port 6379        # Redis



http_access allow devnet

重新部署dify

在dify的的docker目錄中修改docker-compose.yaml文件sandbox使用本地鏡像。

sandbox:

    #image: langgenius/dify-sandbox:0.2.10

    image: dify-sandbox:local

    restart: always

    environment:

將image由langgenius/dify-sandbox:0.2.10 改為了dify-sandbox:local

在docker目錄下執行以下命令

# 銷毀

docker compose down

# 重新部署

docker compose up -d

腳本

使用kimi生成了一個python代碼

import sys

import pymysql

import os



def connect_to_database():

    """ 連接到數據庫，配置都從環境變量里取 """

    try:

        # 從環境變量或配置文件中獲取數據庫參數

        host = os.getenv("DB_HOST", "localhost")

        user = os.getenv("DB_USER", "root")

        password = os.getenv("DB_PASSWORD", "password")

        database = os.getenv("DB_NAME", "database_name")



        conn = pymysql.connect(

            host=host,

            user=user,

            password=password,

            database=database,

            charset='utf8mb4',

            cursorclass=pymysql.cursors.DictCursor  # 使用字典游標

        )

        return conn

    except pymysql.MySQLError as err:

        print(f"Error connecting to database: {err}")

        returnNone



def execute_query(conn, query, params=None):

    """ 執行 SQL 查詢，并支持參數化查詢 """

    cursor = conn.cursor()

    try:

        if params:

            cursor.execute(query, params)

        else:

            cursor.execute(query)

        return cursor.fetchall()

    except pymysql.MySQLError as err:

        print(f"Error executing query: {err}")

        returnNone

    finally:

        cursor.close()



def generate_markdown_table(results):

    """ 生成 Markdown 表格 """

    ifnot results:

        return""

    # 獲取列名

    columns = results[0].keys()



    # 表頭

    table_md = "| " + " | ".join([col for col in columns]) + " |\n"

    # 分隔線

    table_md += "| " + " --- |" * len(columns) + "\n"



    # 表格內容

    for row in results:

        table_md += "| " + " | ".join([str(cell) for cell in row.values()]) + " |\n"



    return table_md



def main(arg1: str) -> dict:

    # 參數化查詢，避免 SQL 注入

    query = "SELECT * FROM table_name WHERE column LIKE %s"

    params = ('%' + arg1 + '%',)



    # 連接到數據庫

    conn = connect_to_database()

    ifnot conn:

        sys.exit(1)



    try:

        # 執行查詢

        result = execute_query(conn, query, params)



        if result isNone:

            return {"result": [], "markdown": ""}



        # 生成 Markdown 表格

        markdown_table = generate_markdown_table(result)



        return {

            "result": result,

            "markdown": markdown_table

        }

    except Exception as e:

        print(f"Unexpected error: {e}")

        return {"result": [], "markdown": ""}

    finally:

        # 確保數據庫連接關閉

        conn.close()

后記

這兩種方式，不管哪種都能實現查詢數據庫，但是有個問題，數據量小的時候性能還行，數據量大了，你查詢一次就得耗時好久。

如果知識固定，也可以前置設置一個知識庫把關鍵詞和文章映射出來，這樣大模型整理的時候，盡量的去往對應的關鍵詞上靠。

本文轉載自公眾號@5ycode

上一篇：

MCP協議詳解：復刻Manus全靠它，為什么說MCP是Agent進化的一大步？

下一篇：

2025年AI大模型排行：從代碼生成到多模態處理，誰是最佳選擇？
#你可能也喜歡這些API文章!

我們有何不同？

API服務商零注冊

多API并行試用

數據驅動選型，提升決策效率

查看全部API→
??

熱門場景實測，選對API

#AI文本生成大模型API

對比大模型API的內容創意新穎性、情感共鳴力、商業轉化潛力

25個渠道
一鍵對比試用API 限時免費

#AI深度推理大模型API

對比大模型API的邏輯推理準確性、分析深度、可視化建議合理性

10個渠道
一鍵對比試用API 限時免費