提升您的生产力：使用 Qdrant 和 Llamaindex 创建自主 AI 工具以实现无缝工作流程

Rifx.Online
Natural Language Processing , AI Applications , AI Research
05 Mar, 2025

Agentic AI：使用 Qdrant 和 LlamaIndex 简化工作流程

架构：

在架构中，Streamlit 充当用户界面，用户可以在其中使用自然语言或通过简单输入提供他们的需求。然后，用户的请求被传递给 Qdrant，它充当一个语义路由器。Qdrant 负责理解用户输入的上下文和语义。基于这种理解，它决定应该触发哪个工作流程。这使得 Qdrant 在确保执行正确的工作流程、优化效率和准确性方面至关重要。

一旦确定了适当的工作流程，LlamaIndex 就会驱动执行。该系统展示了两个核心工作流程。第一个是日历管理工作流程，代理与 Google 日历交互，代表用户创建事件、阻止日历时段和管理日程安排。第二个工作流程支持创建演示文稿，代理获取用户提供的数据，对其进行处理，并以 Microsoft PowerPoint 格式生成结构化演示文稿。LlamaIndex 通过与相关工具和 API 集成来确保这些工作流程得到无缝处理，从而提供所需的输出。

该系统突出了多代理设置在提高生产力方面的巨大潜力。通过减少对日程安排或演示文稿创建等日常任务所需的手动工作，它允许用户专注于高优先级活动。该架构对高级管理人员和管理层特别有利，可以显着提高生产力并简化他们的日常运营。

实现方式：

首先，我们按照以下脚手架格式创建项目。

.
├── calender_agent_main.py
├── calender_agent_workflow.py
├── presentation_creator.iml
├── presentation_generator_agent.py
├── presentation_generator_agent_workflow.py
├── readme.md
└── requirements.txt

“.env” 和 “requirements.txt” 文件如下所示。

OPENAI_API_KEY="sk-proj-"
ANTHROPIC_API_KEY='sk-ant-'
GROQ_API_KEY='gsk_'
COMPOSIO_API_KEY="k60jcx2mj6o335whpdxpoz" #if you have done composio login, you dont need this
STREAMLIT_SERVER_TIMEOUT=300
STREAMLIT_SERVER_MAX_MESSAGE_SIZE=1000

composio-llamaindex
llama-index-llms-openai
llama-index-llms-anthropic
llama-index-llms-groq
llama-index-utils-workflow
python-dotenv
composio-core
streamlit
pytz

Semantic Router 与 Qdrant 的 Fastembed 编码相结合，可以足够快地做出决策，将请求路由到正确的工作流程。

from semantic_router import Route
from semantic_router.layer import RouteLayer
from semantic_router.llms.ollama import OllamaLLM
from semantic_router.encoders import FastEmbedEncoder

calender_workflow = Route(
    name="calender_workflow",
    utterances=[
        "Can you schedule a meeting for tomorrow at 10 AM?",
        "Block my calendar for a team lunch next Friday at noon.",
        "Set up a meeting with the marketing team for Monday at 3 PM.",
        "I need to block my calendar for a doctor's appointment on Thursday at 5 PM.",
        "Schedule a brainstorming session for this Wednesday at 2 PM.",
        "Please block the whole day next Tuesday for a conference.",
        "Book a meeting with John and Sarah for 11 AM tomorrow.",
        "Can you arrange a meeting with the sales team on Friday at 4 PM?",
        "I need my calendar blocked for the next two hours starting now.",
        "Reserve 3 PM this Thursday for a project review meeting.",
        "Schedule an event on my calendar for Saturday evening at 7 PM.",
        "Please block time for a client call next Wednesday at 10 AM.",
        "I need to set up a meeting with the IT team this Friday at 11 AM.",
        "Mark my calendar as busy for lunch with the CEO at 1 PM tomorrow.",
        "Schedule a weekly stand-up meeting every Monday at 9 AM.",
        "Block my calendar for a webinar on artificial intelligence on Thursday at 6 PM.",
        "Reserve the entire afternoon next Wednesday for personal work.",
        "Can you set up a meeting with the HR team for next Monday at 10 AM?",
        "Block off 10 AM to 11:30 AM on Tuesday for a strategy session.",
        "Schedule a one-on-one meeting with my mentor for this Thursday at 4 PM.",
        "Mark 3 PM on Friday for a follow-up call with the client.",
        "Book a recurring meeting for the next four Wednesdays at 2 PM.",
        "Block my calendar for a family event on Sunday at 5 PM.",
        "Schedule a quick sync-up meeting tomorrow at 8 AM."
    ]
)

presentation_workflow = Route(
    name="presentation_workflow",
    utterances=[
        "Can you create a presentation using the sales data from the Google Sheet?",
        "Use the data in the Google Sheet to make a project update presentation.",
        "Please generate a presentation based on the budget report in the shared sheet.",
        "Create slides summarizing the marketing data in the Google Sheet.",
        "Make a presentation with the revenue figures from the spreadsheet.",
        "Can you prepare a PowerPoint using the product launch data from the sheet?",
        "Use the shared Google Sheet to create a summary slide deck.",
        "I need a presentation created using the quarterly performance data in the sheet.",
        "Take the data from the Google Sheet and make a client-ready presentation.",
        "Can you use the chart data from the sheet to build slides for tomorrow's meeting?",
        "Prepare a presentation from the inventory data in the spreadsheet.",
        "Create slides from the Google Sheet for the annual report.",
        "Make a pitch deck using the customer feedback data in the shared sheet.",
        "I need a summary presentation based on the Google Sheet's financial data.",
        "Please convert the data in the spreadsheet into a concise presentation.",
        "Build a presentation using the campaign performance metrics in the Google Sheet.",
        "Can you prepare a slide deck based on the attendance data in the shared sheet?",
        "Turn the Google Sheet data into a presentation for the board meeting.",
        "Use the figures in the shared sheet to design a detailed presentation.",
        "Make a presentation from the task completion data in the spreadsheet.",
        "Create a slide deck from the sales trends shown in the Google Sheet.",
        "Use the spreadsheet data to prepare a presentation for the weekly team meeting.",
        "Can you design slides from the survey results in the Google Sheet?",
        "Generate a presentation based on the revenue analysis data in the shared sheet.",
        "Make a presentation summarizing the milestones in the Google Sheet.",
        "I need slides prepared from the shared Google Sheet's project timeline.",
        "Build a presentation highlighting the top insights from the data in the spreadsheet."
    ]
)

### driver to get the workflow name
def get_route(query: str):
    encoder = FastEmbedEncoder(name="snowflake/snowflake-arctic-embed-s")
    routes = [calender_workflow, presentation_workflow]
    llm = OllamaLLM(llm_name="qwen2.5:latest")

    sr = RouteLayer(encoder=encoder, routes=routes, llm=llm)
    return sr(query).name

PPT 生成器将获取用户查询，并从中提取谷歌表格 ID，最后触发工作流程。下面的方法展示了 LLM 提取 google_sheet_id。

示例提示：

create the presentation by refering to the data source at https://docs.google.com/spreadsheets/d/1JJZdYpyEFsF-IXUa5Ek30wlNdAueICMf26BLUQLnbuU/edit?gid=793403375#gid=793403375

def get_google_sheet_id_from_promot(user_prompt: str):
    system_prompt = """
    As a personal assistant, analyze the given prompt and extract the Google Sheet ID. A Google Sheet ID is a unique
    string of characters that appears in the Google Sheet URL between /d/ and /edit.

Return the information in the following JSON structure, maintaining the exact format:
    {
        "sheet_id": "string"
    }

Note: A valid Google Sheet ID typically:
    - Contains letters, numbers, and special characters like '-' and '_'
    - Is approximately 44 characters long
    - Does not contain spaces or common URL characters like '/' or '?'

Note: dont enclose in back tick like ```. just return plain JSON object.
    """

将预处理后的提示发送给 LLM

    response = openai.chat.completions.create(
        model="gpt-4o-2024-08-06",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ]
    )
    print(response.choices[0].message.content)

## Extract the response
    llm_response = json.loads(response.choices[0].message.content)
    print(llm_response)
    return llm_response

生成演示文稿的工作流程如下。

from llama_index.core.workflow import (
    StartEvent,
    StopEvent,
    Workflow,
    step, Context
)
from composio_llamaindex import ComposioToolSet, Action
from llama_index.core.agent import FunctionCallingAgentWorker
from llama_index.core.llms import ChatMessage
from llama_index.llms.openai import OpenAI
from events import PrefixMessageEvent, AgentEvent
import dotenv
import json
import openai

### Load environment variables from .env file
dotenv.load_dotenv()

class PresentationGenerationWorkflow(Workflow):
    @step
    async def initialize(self, ev: StartEvent, ctx: Context) -> PrefixMessageEvent:
        number_of_slides = ev.number_of_slides | 10

## set the googlesheet id and llm to ctx
        await ctx.set("llm", OpenAI(model="gpt-4o"))

composio_toolset = ComposioToolSet()

tools = composio_toolset.get_tools(actions=[
            Action.CODEINTERPRETER_EXECUTE_CODE,
            Action.CODEINTERPRETER_GET_FILE_CMD,
            Action.CODEINTERPRETER_RUN_TERMINAL_CMD,
            Action.GOOGLESHEETS_BATCH_GET
        ])
        await ctx.set("tools", tools)

prefix_messages = [
            ChatMessage(
                role="system",
                content=(
                    f"""
                    You are an AI assistant specialized in creating PowerPoint presentations using the python-pptx library.
                    Your task is to analyze the Google Sheets data from the provided spreadsheet ID: {ev.google_sheet_id}.
                    Extract key insights and generate relevant charts based on this data.
                    Finally, create a well-structured presentation that includes these charts and any necessary images, ensuring
                    that the formatting is professional and visually appealing. Always create {number_of_slides} slides with more
                    in-depth information covering all aspects of the sheet. When utilizing the Google Sheets tool, only the
                    spreadsheet ID should be passed as input parameters.
                    NOTE: Mostly the user passes small sheets, so try to read the whole sheet at once and not via ranges.
                    """
                )
            )
        ]

task = f"""
        Create a PowerPoint presentation from the Google Sheet: {ev.google_sheet_id}.
        Create a sandbox First retrieve the sheets content, pip install python-pptx using the code interpreter,
        and then use python-pptx. Then write code to create graphs from the data.
        Ensure the presentation is detailed, visually appealing, and contains {number_of_slides} slides.
        Include charts and tables for key insights and ensure proper formatting.
        """

print(task)

await ctx.set("task", task)
        return PrefixMessageEvent(prefix_messages=prefix_messages)

@step
    async def create_agent(self, ev: PrefixMessageEvent, ctx: Context) -> AgentEvent:
        tools = await ctx.get("tools")
        llm = await ctx.get("llm")
        prefix_messages = ev.prefix_messages

agent = FunctionCallingAgentWorker(
            tools=tools,
            llm=llm,
            prefix_messages=prefix_messages,
            max_function_calls=15,
            allow_parallel_tool_calls=False,
            verbose=True
        ).as_agent()

return AgentEvent(agent=agent)

@step
    async def run_agent(self, ev: AgentEvent, ctx: Context) -> StopEvent:
        agent = ev.agent
        task = await ctx.get("task")
        response = agent.chat(task)
        return StopEvent(result=response)

同样，对于日历工作流程，首先 LLm 将从用户提示中提取核心细节，如下所示。

示例提示：

book a slot for a meeting with keshav tomorrow from 3.00 PM to 4.00 PM on AI agents discussion.

def get_details_from_promot(user_prompt: str):
    system_prompt = """
    As a personal assistant, analyze the given prompt and extract the following information:
    - date: Convert any relative or absolute dates to YYYY-MM-DD format
    - timezone: Extract any mentioned timezone example: IST, CST(leave empty if not specified)
    - timeslot: Format time range as "HH:MM AM/PM - HH:MM AM/PM"
    - intent: Determine if the action is create_meeting with xyz person, create_meeting on topic, cancel_meeting, or reschedule_meeting

Return the information in the following JSON structure, maintaining the exact format:
    {
        "date": "YYYY-MM-DD",
        "timezone": "IST",
        "timeslot": "HH:MM AM/PM - HH:MM AM/PM",
        "intent": "meeting with xyz person or meeting on specific topic etc"
    }
    Note: Always get the today's date"""+str(datetime.today())+"""and formulate the target dates based on this.
    """

## Send the preprocessed prompt to the LLM
    response = openai.chat.completions.create(
        model="gpt-4o-2024-08-06",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ]
    )

## Extract the response
    llm_response = json.loads(response.choices[0].message.content)
    print(llm_response)
    return llm_response

传递给工作流程的提示如下。

    response = get_details_from_promot("book a slot for a meeting with keshav tomorrow from 3.00 PM to 4.00 PM ")
    task = f"""
                Book meeting slots according to "{response['timeslot']} -> {response['intent']}".
                Properly label them with the work provided to be don e in that time period.
                Schedule it for today. Today's date is {response['date']} (it's in YYYY-MM-DD format)
                and make the timezone be {response['timezone']}.
                """

日历日程安排工作流程如下。

from llama_index.core.workflow import (
    StartEvent,
    StopEvent,
    Workflow,
    step, Context
)
from composio_llamaindex import App, ComposioToolSet
from llama_index.core.agent import FunctionCallingAgentWorker
from llama_index.core.llms import ChatMessage
from llama_index.llms.openai import OpenAI
from datetime import datetime
from llama_index.utils.workflow import draw_all_possible_flows
from events import PrefixMessageEvent, AgentEvent
import dotenv
import openai
import json

从 .env 文件加载环境变量

dotenv.load_dotenv()

class CalenderAgenticWorkflow(Workflow):
    @step
    async def initialize(self, ev: StartEvent, ctx: Context) -> PrefixMessageEvent:
        # Initialize the LLM
        llm = OpenAI(model="gpt-4o")

        # Initialize the ComposioToolSet
        toolset = ComposioToolSet()

        # Get the RAG tool from the Composio ToolSet
        tools = toolset.get_tools(apps=[App.GOOGLECALENDAR])

        # Define the prefix message for Agent
        prefix_messages = [
            ChatMessage(
                role="system",
                content=(
                    """
                    You are an AI agent responsible for taking actions on Google Calendar on users' behalf.
                    You need to take action on Calendar using Google Calendar APIs. Use correct tools to run APIs from the given tool-set.
                    """
                ),
            )
        ]

        await ctx.set("query", ev.query)
        await ctx.set("tools", tools)
        await ctx.set("llm", llm)

        return PrefixMessageEvent(prefix_messages=prefix_messages)

    @step
    async def create_agent(self, ev: PrefixMessageEvent, ctx: Context) -> AgentEvent:
        tools = await ctx.get("tools")
        llm = await ctx.get("llm")
        prefix_messages = ev.prefix_messages

        # Initialize a FunctionCallingAgentWorker with the tools, LLM, and system messages
        agent = FunctionCallingAgentWorker(
            tools=tools,  # Tools available for the agent to use
            llm=llm,  # Language model for processing requests
            prefix_messages=prefix_messages,  # Initial system messages for context
            max_function_calls=10,  # Maximum number of function calls allowed
            allow_parallel_tool_calls=False,  # Disallow parallel tool calls
            verbose=True,  # Enable verbose output
        ).as_agent()

        return AgentEvent(agent=agent)

    @step
    async def run_agent(self, ev: AgentEvent, ctx: Context) -> StopEvent:
        agent = ev.agent
        query = await ctx.get("query")
        response = agent.chat(query)
        return StopEvent(result=response)

结论：

总而言之，本项目展示了 Agentic AI 系统在重新定义生产力工具方面的巨大潜力。通过利用由 Qdrant 的 FastEmbed 编码器提供支持的语义路由和由 LlamaIndex 提供支持的工作流程，可以以前所未有的效率和精度自动化诸如日历管理和演示文稿创建等常规但必不可少的任务。 Google 日历和 Microsoft PowerPoint 等工具的无缝集成证明了此类系统在实际应用中的实用性。

这种架构是多代理设置增强用户体验、减少手动工作量和提高生产力的有力证明。它强调了 AI 驱动的工作流程简化专业人士（尤其是领导者）运营的潜力，使他们能够专注于战略决策。随着 AI 的不断发展，这样的系统为更智能、更高效的工作场所铺平了道路，为数字生产力解决方案树立了新标准。扩展此类系统的可能性是巨大的，使其成为未来几年创新和采用的令人兴奋的领域。

github repo: https://github.com/pavanjava/cxo_agent