本地準(zhǔn)備

要求：

接口開發(fā)

讓kimi給我生成一個(gè)文章表，并且插入10條數(shù)據(jù)，我們可以告訴kimi，文章長(zhǎng)度多大，這樣內(nèi)容可以豐富些。

直接讓kimi生成一個(gè)暴露接口查數(shù)據(jù)庫(kù)的服務(wù)，有簡(jiǎn)單的優(yōu)化了下,將下面的內(nèi)容放入到server.py文件中

from flask import Flask, request, jsonify

import pymysql



app = Flask(__name__)



# 數(shù)據(jù)庫(kù)配置

DATABASE_CONFIG = {

    'host': '', # 自己的數(shù)據(jù)庫(kù)地址

    'user': '', # 自己數(shù)據(jù)庫(kù)的賬戶

    'password': '', #自己數(shù)據(jù)庫(kù)的密碼

    'db': 'demo', # 自己數(shù)據(jù)庫(kù)的庫(kù)名

    'charset': 'utf8mb4',

    'cursorclass': pymysql.cursors.DictCursor

}



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

def query_database():

    print("接收到請(qǐng)求")

    # 獲取關(guān)鍵字

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

    print("keyword為："+keyword)

    ifnot keyword:

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



    # 參數(shù)化查詢，避免 SQL 注入，修改成自己的庫(kù)

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

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



    try:

        # 建立數(shù)據(jù)庫(kù)連接

        connection = pymysql.connect(**DATABASE_CONFIG)

        with connection.cursor() as cursor:

            # 執(zhí)行查詢

            cursor.execute(query, params)

            result = cursor.fetchall()



        connection.commit()

        connection.close()



        ifnot result:

            return"未查詢到有效數(shù)據(jù)", 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"



    # 表格內(nèi)容

    for row in results:

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



    return table_md



if __name__ == '__main__':

    # 注意這里綁定本機(jī)的內(nèi)容ip，省事點(diǎn)，就0.0.0.0即可。不要綁定127.0.0.1，docke內(nèi)訪問不到

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

啟動(dòng)服務(wù)

python .\server.py

配置工作流

創(chuàng)建一個(gè)空白應(yīng)用。

在開始節(jié)點(diǎn)添加一個(gè)輸入字段context

添加一個(gè)LLM，把開始節(jié)點(diǎn)設(shè)置的context字段作為上下文傳入，并設(shè)置提示詞提取關(guān)鍵詞。

添加一個(gè)http請(qǐng)求節(jié)點(diǎn)，把我們?cè)诮涌陂_發(fā)里的地址和接口名填寫進(jìn)去2，然后把大模型的輸出作為關(guān)鍵詞填寫到請(qǐng)求body里3，我們關(guān)閉重試機(jī)制4。

這里要注意下：json的引號(hào)是中文的，最好在外面寫好校驗(yàn)過了再放進(jìn)去。

在HTTP請(qǐng)求的輸出變量里，我們只關(guān)注status_code 響應(yīng)狀態(tài)碼和響應(yīng)內(nèi)容即可。

添加一個(gè)條件分支1，然后設(shè)置HTTP響應(yīng)碼為200的時(shí)候，連接到大模型。其他直接結(jié)束。

添加大模型，將HTTP請(qǐng)求的響應(yīng)體作為上下文給大模型，輸入提示詞，讓大模型根據(jù)知識(shí)，驗(yàn)證，并進(jìn)行合理性的驗(yàn)證，最后結(jié)構(gòu)化返回。

在結(jié)束節(jié)點(diǎn)中，我們把大模型整理的內(nèi)容輸出。

試運(yùn)行效果。

基于代碼執(zhí)行查詢數(shù)據(jù)庫(kù)

由于difysandbox的安全限制

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

difysandbox源碼修改

一定要使用linux環(huán)境、一定要使用linux環(huán)境一定要使用linux環(huán)境

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

一共有4個(gè)文件，

我用python，不是arm架構(gòu)的，鏡像都是linux的。

我們直接問kimi即可。

ps:這個(gè)問題丟給了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,  

}

預(yù)裝mysql操作包

既然我們要操作在沙箱里操作mysql，那我們得在對(duì)應(yīng)的環(huán)境中預(yù)裝下mysql客戶端。

在1對(duì)應(yīng)的文件中添加2對(duì)應(yīng)的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.

編譯成功以后，打包鏡像。因?yàn)槲覜]有環(huán)境，直接模擬了下創(chuàng)建了一個(gè)main和env目錄

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

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

我在win上打包報(bào)了一堆錯(cuò)，都扔給kimi，一步步的解決。最后成功。

沙箱網(wǎng)咯策略配置

在我們的安裝dify的的時(shí)候，有個(gè)dify/docker/ssrf_proxy目錄，找到squid.conf.template

在這里，你可以設(shè)置允許訪問的網(wǎng)絡(luò)，允許訪問的端口，生產(chǎn)一定要最小權(quán)限

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目錄下執(zhí)行以下命令

# 銷毀

docker compose down

# 重新部署

docker compose up -d

腳本

使用kimi生成了一個(gè)python代碼

import sys

import pymysql

import os



def connect_to_database():

    """ 連接到數(shù)據(jù)庫(kù)，配置都從環(huán)境變量里取 """

    try:

        # 從環(huán)境變量或配置文件中獲取數(shù)據(jù)庫(kù)參數(shù)

        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  # 使用字典游標(biāo)

        )

        return conn

    except pymysql.MySQLError as err:

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

        returnNone



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

    """ 執(zhí)行 SQL 查詢，并支持參數(shù)化查詢 """

    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"



    # 表格內(nèi)容

    for row in results:

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



    return table_md



def main(arg1: str) -> dict:

    # 參數(shù)化查詢，避免 SQL 注入

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

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



    # 連接到數(shù)據(jù)庫(kù)

    conn = connect_to_database()

    ifnot conn:

        sys.exit(1)



    try:

        # 執(zhí)行查詢

        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:

        # 確保數(shù)據(jù)庫(kù)連接關(guān)閉

        conn.close()

后記

這兩種方式，不管哪種都能實(shí)現(xiàn)查詢數(shù)據(jù)庫(kù)，但是有個(gè)問題，數(shù)據(jù)量小的時(shí)候性能還行，數(shù)據(jù)量大了，你查詢一次就得耗時(shí)好久。

如果知識(shí)固定，也可以前置設(shè)置一個(gè)知識(shí)庫(kù)把關(guān)鍵詞和文章映射出來，這樣大模型整理的時(shí)候，盡量的去往對(duì)應(yīng)的關(guān)鍵詞上靠。

本文轉(zhuǎn)載自公眾號(hào)@5ycode

上一篇：

MCP協(xié)議詳解：復(fù)刻Manus全靠它，為什么說MCP是Agent進(jìn)化的一大步？

下一篇：

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

我們有何不同？

API服務(wù)商零注冊(cè)

多API并行試用

數(shù)據(jù)驅(qū)動(dòng)選型，提升決策效率

查看全部API→
??

熱門場(chǎng)景實(shí)測(cè)，選對(duì)API

#AI文本生成大模型API

對(duì)比大模型API的內(nèi)容創(chuàng)意新穎性、情感共鳴力、商業(yè)轉(zhuǎn)化潛力

25個(gè)渠道
一鍵對(duì)比試用API 限時(shí)免費(fèi)

#AI深度推理大模型API

對(duì)比大模型API的邏輯推理準(zhǔn)確性、分析深度、可視化建議合理性

10個(gè)渠道
一鍵對(duì)比試用API 限時(shí)免費(fèi)