用Flask做langchain服务的API接口

“ 之前使用langserve可以特别轻松的封装langchain服务为API，这些API开放了链的各种能力。

有时候我们实际上只是需要更加简单的接口，并且希望能够更加灵活的对接口进行控制。此时直接使用flask及相关框架可能是更好的解决方案。”

本文讲述了如何使用一个利用本地大模型llama3.1实现语言翻译功能的API。

它用flask很轻松的将langchian服务开放为API，结合flasgger给文档添加注释并生成playground，并且使用jsonschema做json数据校验，轻松验证接口参数。

麻雀虽小，但五脏俱全，应该对langchain入门者有较大帮助。

安装依赖

pip install flask flask-restful

pip install flasgger

pip install jsonschema

定义翻译方法

下面的方法需要两个参数，其中：language是翻译目标语言，text是需要翻译的文本。

# 翻译方法

def translate(language,text):

    # 1. 创建提示词模板

    system_template = "Translate the following into {language}:"

    prompt_template = ChatPromptTemplate.from_messages([

        ('system', system_template),

        ('user', '{text}')

    ])



    # 2. 创建本地大模型

    model = OllamaLLM(model="llama3.1")



    # 3. 创建解析器

    parser = StrOutputParser()



    # 4. 创建链

    chain = prompt_template | model | parser



    #5. 调用链

    result = chain.invoke({"language": language,"text":text})



    return result

定义请求的json数据格式

当前比较流行的做法是客户端的请求以json格式提交给服务端。

使用jsonschema可以轻松的对客户端的输入参数进行校验。

# 定义请求数据json的格式。l:language；t:text

schema={

    "type": "object",

    'required': ['l','t'],

    'properties': {

        'l': {'type': 'string',"minLength": 2,"maxLength": 100},

        't': {'type': 'string',"minLength": 2,"maxLength": 1000}

    }

}

关于jsonschema的详细内容，可参见：https://python-jsonschema.readthedocs.io/en/stable/

定义接口和路由

使用`@app.route`可以指定接口路由，接口方法内部的注释可以被`flasgger`渲染生成playground。

#翻译API

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

def trans_api():

    # 以下注释将会被flasgger使用。

    """

    翻译文本。

    ---

    tags:

      - 翻译

    description:

        将文本翻译为目标语言。

    consumes:

        - application/json

    produces:

        - application/json

    parameters:

      - name: query

        in: body

        required: true

        description: json格式。例如：{"l":"简体中文","t":"good morning"}            

    responses:

        code==ok: 

            description: 成功。msg的值为返回的内容。

        code==err: 

            description: 失败。例如：{"code":"err","msg":"抱歉，我不知道。"} 。

    """



    try:

        j = request.get_json()

        validate(instance=j, schema=schema)

        r = translate(j["l"].strip(),j["t"].strip())

        return jsonify({"code":"ok","msg":r})

    except Exception as e:

        return jsonify({"code":"err","msg":str(e)})

关于flasgger详细的说明，请参考：

启动API

if __name__ == '__main__':

    #r = translate("简体中文","good morning")

    #print(r)



    # 设置API 文档。API文档访问地址：http://127.0.0.1:5001/apidocs/

    swagger = Swagger(app=app)



    app.run(port=5001)

出现下图所示，即表示API启动成功。