Chainlit集成Langchain并使用通义千问实现文生图网页应用

前言

本文教程如何使用通义千问的大模型服务平台的接口，实现图片生成的网页应用，主要用到的技术服务有，chainlit 、 langchain、 flux。合利用了大模型的工具选择调用能力。实现聊天对话生成图片的网页应用。
阿里云大模型服务平台百炼 API接口文档地址 https://help.aliyun.com/zh/model-studio/developer-reference/flux-quick-start?spm=a2c4g.11186623.0.0.178f5598SOMvIt ，下面是代码教程。

创建一个文件，例如“chainlit_chat”

mkdir chainlit_chat

进入 chainlit_chat文件夹下，执行命令创建python 虚拟环境空间(需要提前安装好python sdk。 Chainlit 需要python>=3.8。,具体操作，由于文章长度问题就不在叙述，自行百度)，命令如下：

python -m venv .venv

这一步是避免python第三方库冲突，省事版可以跳过
.venv是创建的虚拟空间文件夹可以自定义

接下来激活你创建虚拟空间，命令如下：

#linux or mac
source .venv/bin/activate
#windows
.venv\Scripts\activate

在项目根目录下创建`requirements.txt`，内容如下：

chainlit~=1.1.404
langchain~=0.2.16
langchain_community
stability_sdk==0.8.0
dashscope~=1.20.9
pillow~=10.4.0
requests~=2.32.3

stability_sdk 可以去掉，我实例代码中用的dashscope提供的图片生成接口。

执行以下命令安装依赖：

pip install -r .\requirements.txt

安装后，项目根目录下会多出.chainlit 和.files文件夹和chainlit.md文件

在项目根目录下创建`.env`环境变量，配置如下：

DASHSCOPE_API_KEY="sk-api_key"

DASHSCOPE_API_KEY 是阿里dashscope的服务的APIkey，代码中使用DashScope的sdk实现，所以不需要配置base_url。默认就是阿里的base_url。
阿里模型接口地址 https://dashscope.console.aliyun.com/model

在项目根目录下创建app.py文件，代码如下：

import chainlit as cl
from chainlit.action import Action
from chainlit.input_widget import Select
from langchain.agents import AgentExecutor, AgentType, initialize_agent
from langchain.agents.structured_chat.prompt import SUFFIX
from langchain.memory import ConversationBufferMemory
from langchain_community.llms.tongyi import Tongyi

from tools import edit_image_tool, generate_image_tool


@cl.action_callback("Create variation")
async def create_variant(action: Action):
    agent_input = f"Create a variation of {action.value}"
    await cl.Message(content=f"Creating a variation of `{action.value}`.").send()
    await main(cl.Message(content=agent_input))


@cl.author_rename
def rename(orig_author):
    mapping = {
        "LLMChain": "Assistant",
    }
    return mapping.get(orig_author, orig_author)


@cl.cache
def get_memory():
    return ConversationBufferMemory(memory_key="chat_history")


@cl.on_chat_start
async def start():
    settings = await cl.ChatSettings(
        [
            Select(
                id="image_size",
                label="图片尺寸",
                values=["512*1024", "768*512", "768*1024", "1024*1024"],
                initial_index=1,
                tooltip="Measured in pixels",
            )
        ]
    ).send()
    await setup_agent(settings)


@cl.on_settings_update
async def setup_agent(settings):
    print("Setup agent with following settings: ", settings)
    llm = Tongyi(model='qwen-plus', verbose=True)
    memory = get_memory()
    _SUFFIX = "Chat history:\n{chat_history}\n\n" + SUFFIX

    agent = initialize_agent(
        llm=llm,
        tools=[generate_image_tool, edit_image_tool],
        agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
        memory=memory,
        agent_kwargs={
            "suffix": _SUFFIX,
            "input_variables": ["input", "agent_scratchpad", "chat_history"],
        },
    )
    cl.user_session.set("image_size", settings["image_size"])
    cl.user_session.set("agent", agent)


@cl.on_message
async def main(message: cl.Message):
    agent = cl.user_session.get("agent")  # type: AgentExecutor
    cl.user_session.set("generated_image", None)

    # No async implementation in the Stability AI client, fallback to sync
    res = await cl.make_async(agent.run)(
        input=message.content, callbacks=[cl.LangchainCallbackHandler()]
    )

    elements = []
    actions = []

    generated_image_name = cl.user_session.get("generated_image")
    print('generated_image_name', generated_image_name)
    generated_image = cl.user_session.get(generated_image_name)
    print('generated_image', generated_image)
    if generated_image:
        elements = [
            cl.Image(
                url=generated_image,
                name=generated_image_name,
                display="inline",
            )
        ]
        actions = [cl.Action(name="Create variation", value=generated_image_name)]

    await cl.Message(content=res, elements=elements, actions=actions).send()

db = SQLDatabase.from_uri("sqlite:///demo.db") 中的demo.db文件是上面sqlite_data.py文件执行后生成的
llm = Tongyi(model='qwen-plus', verbose=True) 中verbose 意思是是否打印详细输出
在底层，LangChain 使用 SQLAlchemy 连接到 SQL 数据库。因此，SQLDatabaseChain 可以与 SQLAlchemy 支持的任何 SQL 方言一起使用，例如 MS SQL、MySQL、MariaDB、PostgreSQL、Oracle SQL、Databricks 和 SQLite。有关连接到数据库的要求的更多信息，请参阅 SQLAlchemy 文档。

在项目根目录下创建tool.py文件，代码如下：

from http import HTTPStatus

import chainlit as cl
import dashscope
from langchain.tools import StructuredTool, Tool


def get_image_name():
    image_count = cl.user_session.get("image_count")
    if image_count is None:
        image_count = 0
    else:
        image_count += 1

    cl.user_session.set("image_count", image_count)

    return f"image-{image_count}"


def _generate_image(prompt: str, init_image=None):
    size = cl.user_session.get("image_size", '768*512')
    rsp = dashscope.ImageSynthesis.call(model="flux-schnell",
                                        prompt=prompt,
                                        ref_img=init_image,
                                        size=size)
    if rsp.status_code == HTTPStatus.OK:
        result = rsp.output.results[0]
        cl.user_session.set("generated_image", result.url)
        name = get_image_name()
        cl.user_session.set(name, result.url)
        cl.user_session.set("generated_image", name)
        return name
    else:
        print('Failed, status_code: %s, code: %s, message: %s' %
              (rsp.status_code, rsp.code, rsp.message))
        return rsp.message

def generate_image(prompt: str):
    image_name = _generate_image(prompt)
    return f"Here is {image_name}."


def edit_image(init_image_name: str, prompt: str):
    init_image_url = cl.user_session.get(init_image_name)
    if init_image_url is None:
        raise ValueError(f"Could not find image `{init_image_name}`.")

    image_name = _generate_image(prompt, init_image_url)

    return f"Here is {image_name} based on {init_image_name}."


generate_image_tool = Tool.from_function(
    func=generate_image,
    name="GenerateImage",
    description="Useful to create an image from a text prompt.",
    return_direct=True,
)

edit_image_tool = StructuredTool.from_function(
    func=edit_image,
    name="EditImage",
    description="Useful to edit an image with a prompt. Works well with commands such as 'replace', 'add', 'change', 'remove'.",
    return_direct=True,
)