Chainlit集成LlamaIndex实现一个通过用户聊天对话的酒店预定系统

Agent 简介

“Agent”是一个自动推理和决策引擎。它接受用户输入/查询，并为执行该查询做出内部决策，以便返回正确的结果。关键的代理组件可以包括但不限于：

• 把复杂的问题分解成小问题
• 选择要使用的外部工具+调用工具的参数
• 计划一系列的任务
• 将以前完成的任务存储在内存模块中

LlamaIndex为构建不同复杂程度的代理系统提供了一个全面的框架：

• 如果您想快速构建代理：使用LlamaIndex的预构建代理和工具架构来快速设置代理系统。
• 如果您想完全控制您的代理系统：使用LlamaIndex的工作流从头开始构建和部署自定义代理工作流。

LlamaIndex官方地址 https://docs.llamaindex.ai/en/stable/

快速上手

创建一个文件，例如“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
llama-index-core
llama-index-llms-dashscope
llama-index-embeddings-dashscope

执行以下命令安装依赖：

pip install -r .\requirements.txt

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

项目简介

通过使用Chainlit集成LlamaIndex实现，AI在和用户聊天的过程中，依据用户的对话中输入内容，提起用户的关键信息，调用创建预定信息、修改预定信息、验证预定信息、查询预定信息等方法，将用户的信息，转换成代码中预定信息对象，存到内存或数据库中，适用大多数预定场景，不止酒店预定，还可以理发预约、美容预约、按摩预约、商品预定等场景。

代码创建

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

方式1（高级接口实现）

import os
import time
from typing import Optional

import chainlit as cl
from llama_index.core import (
    Settings, )
from llama_index.core.agent import ReActAgent
from llama_index.core.bridge.pydantic import BaseModel
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.tools import FunctionTool
from llama_index.embeddings.dashscope import DashScopeEmbedding, DashScopeTextEmbeddingModels, \
    DashScopeTextEmbeddingType
from llama_index.llms.dashscope import DashScope, DashScopeGenerationModels

Settings.embed_model = DashScopeEmbedding(
    model_name=DashScopeTextEmbeddingModels.TEXT_EMBEDDING_V2,
    text_type=DashScopeTextEmbeddingType.TEXT_TYPE_DOCUMENT,
)
Settings.node_parser = SentenceSplitter(chunk_size=128, chunk_overlap=20)
Settings.num_output = 512
Settings.context_window = 6000

# 我们将在随机id下存储预订
bookings = {}


class Booking(BaseModel):
    name: Optional[str] = None
    phone: Optional[str] = None
    date: Optional[str] = None


user_id = "user123"


def get_booking_state() -> str:
    global user_id
    """获取给定预订ID的预订的当前状态。"""
    try:
        return str(bookings[user_id].dict())
    except:
        return f"未找到 {user_id} 的预定信息"


def update_booking(property: str, value: str) -> str:
    global user_id
    """更新给定预订ID的预订属性。只输入明确提供的详细信息。"""
    booking = bookings[user_id]
    setattr(booking, property, value)
    return f"用户 {user_id} 预定信息 {property} 更新为 {value}"


def create_booking() -> str:
    global user_id
    """创建一个新的预订并返回预订ID。"""
    bookings[user_id] = Booking()
    return f"用户 {user_id} 已创建预订，但尚未确认。请提供您的姓名、电话、预定日期。"


def confirm_booking() -> str:
    global user_id
    """确认给定预订ID的预订。"""
    booking = bookings[user_id]

    if booking.name is None:
        raise ValueError("请提供你的姓名")

    if booking.phone is None:
        raise ValueError("请提供你的手机号")

    if booking.date is None:
        raise ValueError("请提供你的预定日期")

    return f"预定用户 {user_id} 预定信息已确认!"


get_booking_state_tool = FunctionTool.from_defaults(fn=get_booking_state)
update_booking_tool = FunctionTool.from_defaults(fn=update_booking)
create_booking_tool = FunctionTool.from_defaults(
    fn=create_booking, return_direct=True
)
confirm_booking_tool = FunctionTool.from_defaults(
    fn=confirm_booking, return_direct=True
)

tools = [get_booking_state_tool, update_booking_tool, create_booking_tool, confirm_booking_tool]


@cl.on_chat_start
async def start():
    await cl.Message(
        author="Assistant", content="你好! 我是泰山AI智能助手. 有什么可以帮助你的吗?"
    ).send()


llm = DashScope(
    model_name=DashScopeGenerationModels.QWEN_PLUS,
    api_key=os.environ["DASHSCOPE_API_KEY"],
    max_tokens=512,
    temperature=0.2,
    top_k=20
)


agent = ReActAgent.from_tools(tools, llm=llm, verbose=True)

@cl.on_message
async def main(message: cl.Message):
    start_time = time.time()
    msg = cl.Message(content="", author="Assistant")
    res = agent.chat(message.content)
    await msg.stream_token(res.response)
    print(f"代码执行时间: {time.time() - start_time} 秒")
    await msg.send()

• 该Agent智能,有一定的思考能力，我在代码中测试，目前只有在非流式响应下，正常思考回答。流式响应，问题比较多，经常Agent还没思考用哪些工具呢，就已经有回答返回了！
• 代码中使用的通义千文的大模型，测试了qwen-turbo、qwen-plus、qwen-max，发现qwen-turbo的智能程度不够，不能按照预期返回结果，但是指令遵循的忠实度比较高，qwen-turbo、qwen-plus 的智能程度够了，但是指令遵循的忠实度相对较低，需要调整temperature和top_k等参数到合适的值，才能有不错的效果
• 代码中方法的注释最好用英文写，AI更容易理解和遵循指令

方式2（低级接口实现）

import os
import time
from typing import Optional

import chainlit as cl
from llama_index.core import (
    Settings, )
from llama_index.core.agent import ReActAgent, ReActChatFormatter, ReActOutputParser
from llama_index.core.agent.react.output_parser import extract_final_response, extract_tool_use, action_input_parser
from llama_index.core.agent.react.types import BaseReasoningStep, ResponseReasoningStep, ActionReasoningStep
from llama_index.core.bridge.pydantic import BaseModel
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.output_parsers.utils import extract_json_str
from llama_index.core.tools import FunctionTool
from llama_index.embeddings.dashscope import DashScopeEmbedding, DashScopeTextEmbeddingModels, \
    DashScopeTextEmbeddingType
from llama_index.llms.dashscope import DashScope, DashScopeGenerationModels

Settings.embed_model = DashScopeEmbedding(
    model_name=DashScopeTextEmbeddingModels.TEXT_EMBEDDING_V2,
    text_type=DashScopeTextEmbeddingType.TEXT_TYPE_DOCUMENT,
)
Settings.node_parser = SentenceSplitter(chunk_size=128, chunk_overlap=20)
Settings.num_output = 512
Settings.context_window = 6000

# 我们将在随机id下存储预订
bookings = {}


class Booking(BaseModel):
    name: Optional[str] = None
    phone: Optional[str] = None
    date: Optional[str] = None


user_id = "user123"

def get_booking_state() -> str:
    global user_id
    """获取给定预订ID的预订的当前状态。"""
    try:
        return str(bookings[user_id].dict())
    except:
        return f"未找到 {user_id} 的预定信息"


def update_booking(property: str, value: str) -> str:
    global user_id
    """更新给定预订ID的预订属性。只输入明确提供的详细信息。"""
    booking = bookings[user_id]
    setattr(booking, property, value)
    return f"用户 {user_id} 预定信息 {property} 更新为 {value}"


def create_booking() -> str:
    global user_id
    """创建一个新的预订并返回预订ID。"""
    bookings[user_id] = Booking()
    return f"用户 {user_id} 已创建预订，但尚未确认。请提供您的姓名、电话、预定日期。"


def confirm_booking() -> str:
    global user_id
    """确认给定预订ID的预订。"""
    booking = bookings[user_id]

    if booking.name is None:
        raise ValueError("请提供你的姓名")

    if booking.phone is None:
        raise ValueError("请提供你的手机号")

    if booking.date is None:
        raise ValueError("请提供你的预定日期")

    return f"预定用户 {user_id} 预定信息已确认!"

get_booking_state_tool = FunctionTool.from_defaults(fn=get_booking_state)
update_booking_tool = FunctionTool.from_defaults(fn=update_booking)
create_booking_tool = FunctionTool.from_defaults(
    fn=create_booking, return_direct=True
)
confirm_booking_tool = FunctionTool.from_defaults(
    fn=confirm_booking, return_direct=True
)

tools = [get_booking_state_tool, update_booking_tool, create_booking_tool, confirm_booking_tool]


@cl.on_chat_start
async def start():
    await cl.Message(
        author="Assistant", content="你好! 我是泰山AI智能助手. 有什么可以帮助你的吗?"
    ).send()


llm = DashScope(
    model_name=DashScopeGenerationModels.QWEN_PLUS,
    api_key=os.environ["DASHSCOPE_API_KEY"],
    max_tokens=512,
    temperature=0.2,
    top_k=20
)

system_header = """
You are designed to help with a variety of tasks, from answering questions to providing summaries to other types of analyses.
You are now connected to the booking system and helping user with making a booking. 
Only enter details that the user has explicitly provided.
Do not make up any details.

## Tools

You have access to a wide variety of tools. You are responsible for using the tools in any sequence you deem appropriate to complete the task at hand.
This may require breaking the task into subtasks and using different tools to complete each subtask.

You have access to the following tools:
{tool_desc}

## Output Format

Please answer in the same language as the question and use the following format:

Thought: The current language of the user is: (user’s language). I need to use a tool to help me answer the question.
Action: tool name (one of {tool_names}) if using a tool.
Action Input: the input to the tool, in a JSON format representing the kwargs (e.g. {{“input”: “hello world”, “num_beams”: 5}})

Please ALWAYS start with a Thought.

NEVER surround your response with markdown code markers. You may use code markers within your response if you need to.

Please use a valid JSON format for the Action Input. Do NOT do this {{'input': 'hello world', 'num_beams': 5}}.

If this format is used, the user will respond in the following format:

Observation: tool response

## Current Conversation

Below is the current conversation consisting of interleaving human and assistant messages.

"""

react_chat_formatter = ReActChatFormatter.from_defaults(system_header)


def parse_action_reasoning_step(output: str) -> ActionReasoningStep:
    """
    Parse an action reasoning step from the LLM output.
    """
    # Weaker LLMs may generate ReActAgent steps whose Action Input are horrible JSON strings.
    # `dirtyjson` is more lenient than `json` in parsing JSON strings.
    import dirtyjson as json
    thought, action, action_input = extract_tool_use(output)
    json_str = extract_json_str(action_input)
    # First we try json, if this fails we use ast
    try:
        action_input_dict = json.loads(json_str)
    except Exception:
        action_input_dict = action_input_parser(json_str)
    return ActionReasoningStep(
        thought=thought, action=action, action_input=action_input_dict
    )


class CustomReActOutputParser(ReActOutputParser):
    def parse(self, output: str, is_streaming: bool = False) -> BaseReasoningStep:
        """Parse output from ReAct agent.

        We expect the output to be in one of the following formats:
        1. If the agent need to use a tool to answer the question:
            ```
            Thought: <thought>
            Action: <action>
            Action Input: <action_input>
            ```
        2. If the agent can answer the question without any tools:
            ```
            Thought: <thought>
            Answer: <answer>
            ```
        """
        print('CustomReActOutputParser output----------------------start')
        print(output)
        print('CustomReActOutputParser output----------------------end')
        if "Thought:" not in output:
            # NOTE: handle the case where the agent directly outputs the answer
            # instead of following the thought-answer format
            return ResponseReasoningStep(
                thought="(Implicit) I can answer without any more tools!",
                response=output,
                is_streaming=is_streaming,
            )

        # An "Action" should take priority over an "Answer"
        if "Action:" in output:
            return parse_action_reasoning_step(output)

        if "Answer:" in output:
            thought, answer = extract_final_response(output)
            return ResponseReasoningStep(
                thought=thought, response=answer, is_streaming=is_streaming
            )

        raise ValueError(f"Could not parse output: {output}")


output_parser = CustomReActOutputParser()
agent = ReActAgent.from_tools(tools, llm=llm,
                              react_chat_formatter=react_chat_formatter,
                              output_parser=output_parser,
                              verbose=True)



@cl.on_message
async def main(message: cl.Message):
    start_time = time.time()
    msg = cl.Message(content="", author="Assistant")
    task = agent.create_task(message.content)
    print('tasks len: ', len(agent.list_tasks()))
    # 使用while循环直到任务完成
    while True:
        # 如果是第一次或者上一步不是最后一步，则继续执行下一步
        step_output = agent.run_step(task.task_id)
        # 检查是否是最后一步
        if step_output.is_last:
            res = agent.finalize_response(task.task_id)
            await msg.stream_token(res.response)
            break  # 结束循环
    agent.state.reset()
    print(f"代码执行时间: {time.time() - start_time} 秒")
    await msg.send()

• 该代码可以自定义系统提示词、自定义响应解析器

运行应用程序

要启动 Chainlit 应用程序，请打开终端并导航到包含的目录app.py。然后运行以下命令：

 chainlit run app.py -w   

• 该-w标志告知 Chainlit 启用自动重新加载，因此您无需在每次更改应用程序时重新启动服务器。您的聊天机器人 UI 现在应该可以通过http://localhost:8000访问。
• 自定义端口可以追加--port 80

启动后界面如下：

实测下来，发现还有有些回答，不是很符合预期，考虑换其他厂家的API接口或者再调整下提示词和参数试试！

相关文章推荐

《Chainlit快速实现AI对话应用的界面定制化教程》
《Chainlit接入FastGpt接口快速实现自定义用户聊天界面》
《使用 Xinference 部署本地模型》
《Fastgpt接入Whisper本地模型实现语音输入》
《Fastgpt部署和接入使用重排模型bge-reranker》
《Fastgpt部署接入 M3E和chatglm2-m3e文本向量模型》
《Fastgpt 无法启动或启动后无法正常使用的讨论（启动失败、用户未注册等问题这里）》
《vllm推理服务兼容openai服务API》
《vLLM模型推理引擎参数大全》
《解决vllm推理框架内在开启多显卡时报错问题》
《Ollama 在本地快速部署大型语言模型，可进行定制并创建属于您自己的模型》