diff --git a/alias/src/alias/agent/agents/__init__.py b/alias/src/alias/agent/agents/__init__.py index f63ae89d..dfb48fed 100644 --- a/alias/src/alias/agent/agents/__init__.py +++ b/alias/src/alias/agent/agents/__init__.py @@ -1,5 +1,6 @@ # -*- coding: utf-8 -*- from alias.agent.agents._alias_agent_base import AliasAgentBase +from alias.agent.agents._qa_agent import QAAgent from alias.agent.agents._meta_planner import MetaPlanner from alias.agent.agents._browser_agent import BrowserAgent from alias.agent.agents._react_worker import ReActWorker @@ -19,6 +20,7 @@ "ReActWorker", "DeepResearchAgent", "DataScienceAgent", + "QAAgent", "init_ds_toolkit", "init_dr_toolkit", ] diff --git a/alias/src/alias/agent/agents/_alias_agent_base.py b/alias/src/alias/agent/agents/_alias_agent_base.py index 25ca1df6..c24abe51 100644 --- a/alias/src/alias/agent/agents/_alias_agent_base.py +++ b/alias/src/alias/agent/agents/_alias_agent_base.py @@ -3,7 +3,7 @@ import json import time import traceback -from typing import Any, Optional, Literal +from typing import Any, List, Optional, Literal from loguru import logger @@ -24,6 +24,117 @@ class AliasAgentBase(ReActAgent): + @classmethod + async def create( + cls, + name: str, + model: str = "qwen3-max", + system_prompt: Optional[str] = None, + tools: Optional[List[str]] = None, + worker_full_toolkit: Optional[AliasToolkit] = None, + use_long_term_memory_service: bool = False, + ) -> "AliasAgentBase": + """ + Create an AliasAgentBase instance with default configuration. + + This is a convenience factory method that sets up the agent with + appropriate defaults. Tools are registered via share_tools from + worker_full_toolkit when both tools and worker_full_toolkit are + provided. + + Args: + name: The unique identifier name for the agent instance. + model: The model name (e.g., "qwen3-max", "qwen-vl-max"). + Must be a key in MODEL_FORMATTER_MAPPING from run.py. + system_prompt: The system prompt. If None, uses default prompt. + tools: List of tool names to register in the agent's toolkit. + Used together with worker_full_toolkit; tools are copied + via share_tools. + worker_full_toolkit: Source toolkit containing all tools. + When provided together with tools, the specified tools are shared + into the agent's toolkit. + use_long_term_memory_service: Whether to enable long-term memory + service. + + Returns: + A configured AliasAgentBase instance. + """ + from agentscope.memory import InMemoryMemory + from alias.agent.mock import MockSessionService + from alias.agent.run import MODEL_FORMATTER_MAPPING + from alias.agent.tools.share_tools import share_tools + from datetime import datetime + + time_str = datetime.now().strftime("%Y%m%d%H%M%S") + if model not in MODEL_FORMATTER_MAPPING: + raise ValueError( + f"Unknown model name: {model}. " + f"Available models: {list(MODEL_FORMATTER_MAPPING.keys())}", + ) + + model_instance, formatter = MODEL_FORMATTER_MAPPING[model] + session_service = MockSessionService( + use_long_term_memory_service=use_long_term_memory_service, + ) + + # Initialize long-term memory if enabled (same as run.py) + long_term_memory = None + if use_long_term_memory_service: + from alias.server.clients.memory_client import MemoryClient + from alias.agent.memory.longterm_memory import AliasLongTermMemory + + if await MemoryClient.is_available(): + long_term_memory = AliasLongTermMemory( + session_service=session_service, + ) + logger.info( + "Long-term memory service is available and initialized", + ) + else: + logger.warning( + "use_long_term_memory_service is True, but memory " + "service is not available. Long-term memory will not " + "be used. Please check if the memory service is " + "running.", + ) + + # Build toolkit: use worker_full_toolkit's sandbox if provided, + # then share_tools + if worker_full_toolkit is not None: + toolkit = AliasToolkit( + sandbox=worker_full_toolkit.sandbox, + add_all=False, + ) + if tools: + # Validate that each requested tool exists + # in worker_full_toolkit + for tool_name in tools: + if tool_name not in worker_full_toolkit.tools: + raise ValueError( + f"Tool '{tool_name}' is not available. ", + ) + share_tools(worker_full_toolkit, toolkit, tools) + logger.info(f"Shared tools into agent toolkit: {tools}") + else: + toolkit = AliasToolkit(sandbox=None, add_all=False) + + # Create agent instance + agent = cls( + name=name, + model=model_instance, + formatter=formatter, + memory=InMemoryMemory(), + toolkit=toolkit, + session_service=session_service, + state_saving_dir=f"./agent-states/run-{time_str}", + sys_prompt=system_prompt, + max_iters=10, + long_term_memory=long_term_memory, + long_term_memory_mode="both", + ) + + return agent + def __init__( self, name: str, diff --git a/alias/src/alias/agent/agents/_qa_agent.py b/alias/src/alias/agent/agents/_qa_agent.py new file mode 100644 index 00000000..e8cde963 --- /dev/null +++ b/alias/src/alias/agent/agents/_qa_agent.py @@ -0,0 +1,387 @@ +# -*- coding: utf-8 -*- +""" +QAAgent: A specialized agent for question answering with RAG capabilities. + +This agent extends AliasAgentBase to provide GitHub MCP tools and +RAG (Retrieval-Augmented Generation) for a knowledge base in Qdrant. +""" +import hashlib +import os +from pathlib import Path +from typing import TYPE_CHECKING, List, Optional, Sequence, Union + +from loguru import logger +from agentscope.embedding import DashScopeTextEmbedding +from agentscope.message import TextBlock +from agentscope.mcp import HttpStatelessClient +from agentscope.rag import Document, SimpleKnowledge, QdrantStore, TextReader +from agentscope.rag._document import DocMetadata +from agentscope.tool import execute_shell_command + +from alias.agent.agents._alias_agent_base import AliasAgentBase +from alias.agent.agents.qa_agent_utils.create_rag_file import ( + check_container_running, + collection_exists, + start_qdrant_container, + split_faq_records, +) + +if TYPE_CHECKING: + from alias.agent.tools import AliasToolkit + +# Qdrant configuration +QDRANT_HOST = "127.0.0.1" +QDRANT_PORT = 6333 +QDRANT_CONTAINER_NAME = "qdrant" + +# Default RAG file and collection when user does not specify +DEFAULT_RAG_FILE_PATH = ( + Path(__file__).parent / "qa_agent_utils" / "as_faq_samples.txt" +) +DEFAULT_COLLECTION_NAME = "as_faq" + + +class QAAgent(AliasAgentBase): + """QA Agent with RAG capabilities for question answering.""" + + @staticmethod + def _get_default_system_prompt(name: str) -> str: + """ + Get the default system prompt for QAAgent. + + Args: + name: The agent's name. + + Returns: + Default system prompt string. + """ + try: + # Try to load from the built-in prompt file + prompt_file = ( + Path(__file__).parent + / "qa_agent_utils" + / "build_in_prompt" + / "qaagent_base_sys_prompt.md" + ) + if prompt_file.exists(): + prompt = prompt_file.read_text(encoding="utf-8") + return prompt.format(name=name) + except Exception as e: + logger.warning(f"Could not load default QA prompt: {e}") + + # Fallback to a simple default prompt + return ( + f"You are a helpful assistant named {name}.\n\n" + "**IMPORTANT**: You MUST use the `retrieve_knowledge` tool to " + "search the knowledge base FIRST before answering. " + "Do not answer from training data alone if the question may be in " + "the knowledge base.\n\n" + "The `query` parameter is crucial for retrieval quality. " + "Try multiple queries; adjust `limit` and `score_threshold` " + "for number and relevance of results.\n\n" + ) + + @classmethod + async def create( # pylint: disable=too-many-branches,too-many-statements + cls, + name: str, + model: str = "qwen3-max", + system_prompt: Optional[str] = None, + tools: Optional[List[str]] = None, + worker_full_toolkit: Optional["AliasToolkit"] = None, + use_long_term_memory_service: bool = False, + file: Optional[List[Union[str, Path]]] = None, + collection_name: Optional[str] = None, + ) -> "QAAgent": + """ + Create a QAAgent instance with RAG capabilities. + + Args: + name: The unique identifier name for the agent instance. + model: The model name (e.g., "qwen3-max", "qwen-vl-max"). + system_prompt: The system prompt. If None, uses default prompt. + tools: Tool names to register from worker_full_toolkit. + worker_full_toolkit: Optional. If provided, use this toolkit (same + sandbox/share_tools as AliasAgentBase). If None, create + sandbox and full toolkit internally. + use_long_term_memory_service: Whether to enable long-term memory. + file: List of file paths to process. None to use default or skip. + collection_name: Qdrant collection. None = default 'as_faq'. + + Returns: + A configured QAAgent instance with RAG capabilities. + """ + # Validate inputs + if file is not None and not isinstance(file, list): + raise ValueError("file must be a list of file paths or None") + + # Resolve collection_name (RAG tool uses this collection) + coll_name = ( + collection_name + if collection_name is not None + else DEFAULT_COLLECTION_NAME + ) + + qdrant_running = check_container_running(QDRANT_CONTAINER_NAME) + + if not qdrant_running: + # RAG not initialized: start Qdrant, init (file, collection) + try: + start_qdrant_container() + except Exception as e: + logger.warning(f"Could not start Qdrant container: {e}") + logger.warning("RAG functionality may not work properly") + else: + # Resolve (files to process, collection) for initial load + if file is None and collection_name is None: + files_to_process = [DEFAULT_RAG_FILE_PATH] + init_collection = DEFAULT_COLLECTION_NAME + elif file is not None and collection_name is None: + files_to_process = file + init_collection = DEFAULT_COLLECTION_NAME + elif file is None and collection_name is not None: + files_to_process = [DEFAULT_RAG_FILE_PATH] + init_collection = collection_name + else: + files_to_process = file + init_collection = collection_name + await cls._process_files(files_to_process, init_collection) + else: + # Qdrant running: collection_name is the one this agent uses + if file: + await cls._process_files(file, coll_name) + elif not collection_exists(coll_name): + logger.info( + f"Collection '{coll_name}' does not exist; " + "using default file to populate.", + ) + if DEFAULT_RAG_FILE_PATH.exists(): + await cls._process_files( + [DEFAULT_RAG_FILE_PATH], + coll_name, + ) + else: + logger.warning( + f"Default RAG file not found: {DEFAULT_RAG_FILE_PATH}", + ) + + # Use default system prompt if not provided + if system_prompt is None: + system_prompt = cls._get_default_system_prompt(name) + + # Use worker_full_toolkit or build sandbox + toolkit internally + if worker_full_toolkit is None: + try: + from alias.runtime.alias_sandbox.alias_sandbox import ( + AliasSandbox, + ) + from alias.agent.tools import AliasToolkit + from alias.agent.tools.add_tools import add_tools + + sandbox = AliasSandbox() + sandbox.__enter__() + worker_full_toolkit = AliasToolkit(sandbox, add_all=True) + try: + await add_tools(worker_full_toolkit) + except Exception as e: + logger.warning( + f"add_tools failed: {e}; " + "continuing with sandbox tools only", + ) + logger.info("Created sandbox and full toolkit for QAAgent") + except Exception as e: + logger.warning(f"Could not create sandbox for QAAgent: {e}") + worker_full_toolkit = None + + # Create agent using parent's create (tools + worker_full_toolkit) + agent = await super().create( + name=name, + model=model, + system_prompt=system_prompt, + tools=tools or [], + worker_full_toolkit=worker_full_toolkit, + use_long_term_memory_service=use_long_term_memory_service, + ) + + # Register RAG and GitHub tools on top of shared toolkit + await cls._register_rag_tool(agent, coll_name) + await cls._register_github_tools(agent) + + return agent + + @staticmethod + async def _process_files( + file_paths: Sequence[Union[str, Path]], + collection_name: str, + ) -> None: + """ + Process files and add them to the Qdrant collection. + + Args: + file_paths: List of file paths to process. + collection_name: Name of the Qdrant collection to add documents to. + """ + logger.info( + f"Processing {len(file_paths)} file(s) " + f"for collection '{collection_name}'", + ) + + # Create knowledge base instance + knowledge = SimpleKnowledge( + embedding_store=QdrantStore( + location=None, + client_kwargs={ + "host": QDRANT_HOST, + "port": QDRANT_PORT, + }, + collection_name=collection_name, + dimensions=1024, + ), + embedding_model=DashScopeTextEmbedding( + api_key=os.environ.get("DASHSCOPE_API_KEY"), + model_name="text-embedding-v4", + ), + ) + + # Process each file + reader = TextReader(chunk_size=2048, split_by="char") + all_documents = [] + + for file_path in file_paths: + file_path = Path(file_path) + if not file_path.exists(): + logger.warning(f"File not found: {file_path}, skipping...") + continue + + logger.info(f"Processing file: {file_path}") + try: + with open(file_path, "r", encoding="utf-8") as f: + full_text = f.read() + except Exception as e: + logger.error(f"Error reading file {file_path}: {e}") + continue + + # Split by FAQ records if applicable, otherwise use full text + faq_records = split_faq_records(full_text) + + for faq_record in faq_records: + # If the record is short enough, use it as-is + if len(faq_record) <= 2048: + doc_id = hashlib.sha256( + faq_record.encode("utf-8"), + ).hexdigest() + all_documents.append( + Document( + id=doc_id, + metadata=DocMetadata( + content=TextBlock( + type="text", + text=faq_record, + ), + doc_id=doc_id, + chunk_id=0, + total_chunks=1, + ), + ), + ) + else: + # If too long, split it further using TextReader + chunked_docs = await reader(text=faq_record) + all_documents.extend(chunked_docs) + + if all_documents: + await knowledge.add_documents(all_documents) + logger.info( + f"Successfully added {len(all_documents)} document(s) " + f"to collection '{collection_name}'", + ) + else: + logger.warning( + "No documents were processed from the provided files", + ) + + @staticmethod + async def _register_rag_tool( + agent: "QAAgent", + collection_name: str, + ) -> None: + """ + Register the retrieve_knowledge tool for RAG. + + Args: + agent: The agent instance to register the tool for. + collection_name: Name of the Qdrant collection to use. + """ + import traceback + + try: + knowledge = SimpleKnowledge( + embedding_store=QdrantStore( + location=None, + client_kwargs={ + "host": QDRANT_HOST, # Qdrant server address + "port": QDRANT_PORT, # Qdrant server port + }, + collection_name=collection_name, + dimensions=1024, # The dimension of the embedding vectors + ), + embedding_model=DashScopeTextEmbedding( + api_key=os.environ["DASHSCOPE_API_KEY"], + model_name="text-embedding-v4", + ), + ) + agent.toolkit.register_tool_function( + knowledge.retrieve_knowledge, + func_description=( + "Quickly retrieve answers from the knowledge base. " + "The `query` parameter is crucial for retrieval quality. " + "Try multiple queries; adjust `limit` and " + "`score_threshold` for relevance of results." + ), + ) + logger.info( + f"Registered retrieve_knowledge tool " + f"with collection '{collection_name}'", + ) + except Exception as e: + print(traceback.format_exc()) + raise e from None + + @staticmethod + async def _register_github_tools(agent: "QAAgent") -> None: + """ + Register GitHub MCP tools for the QA agent. + + Args: + agent: The agent instance to register the tools for. + """ + import traceback + + github_token = os.getenv("GITHUB_TOKEN") + if not github_token: + logger.error( + "Missing GITHUB_TOKEN; GitHub MCP tools cannot be used. " + "Please export GITHUB_TOKEN in your environment.", + ) + else: + try: + github_client = HttpStatelessClient( + name="github", + transport="streamable_http", + url="https://api.githubcopilot.com/mcp/", + headers={"Authorization": (f"Bearer {github_token}")}, + ) + + await agent.toolkit.register_mcp_client( + github_client, + enable_funcs=[ + "search_repositories", + "search_code", + "get_file_contents", + ], + ) + agent.toolkit.register_tool_function(execute_shell_command) + logger.info("Registered GitHub MCP tools") + except Exception as e: + print(traceback.format_exc()) + raise e from None diff --git a/alias/src/alias/agent/agents/create_agent.py b/alias/src/alias/agent/agents/create_agent.py new file mode 100644 index 00000000..6ca43832 --- /dev/null +++ b/alias/src/alias/agent/agents/create_agent.py @@ -0,0 +1,303 @@ +# -*- coding: utf-8 -*- +""" +Create a qa agent with name, system_prompt, tools, model, file and +collection_name. + +Example: + python -m alias.agent.agents.create_agent -n QA -a qaagent + --task "What's agentscope?" +""" +import argparse +import asyncio +import os +import sys +import traceback +from pathlib import Path +from typing import List, Optional, Union + +from agentscope.message import Msg +from alias.agent.agents import AliasAgentBase, QAAgent +from alias.agent.tools import AliasToolkit +from alias.agent.tools.add_tools import add_tools +from alias.runtime.alias_sandbox.alias_sandbox import AliasSandbox + + +def _ensure_env(): + """Optional .env for DASHSCOPE_API_KEY. Returns (path or None, created).""" + cwd = Path(os.getcwd()).resolve() + for _ in range(4): + p = cwd / ".env" + if p.exists(): + return None, False + if cwd.parent == cwd: + break + cwd = cwd.parent + p = Path(os.getcwd()) / ".env" + if not p.exists(): + try: + p.write_text( + "ENVIRONMENT=local\nDASHSCOPE_API_KEY=your_key_here\n", + ) + return p, True + except Exception: + pass + return None, False + + +def normalize_agent_type(agent: str) -> str: + """Normalize agent type: qaagent/QAAgent/QA_Agent + -> 'qaagent'; else 'alias'.""" + if not agent or not agent.strip(): + return "alias" + t = agent.strip().lower().replace("_", "").replace("-", "") + return "qaagent" if t == "qaagent" else "alias" + + +def resolve_system_prompt(system_prompt: Optional[str]) -> str: + """Path -> file content; else as-is. None/empty -> ''.""" + if not system_prompt or not system_prompt.strip(): + return system_prompt or "" + p = Path(system_prompt.strip()) + if p.is_file(): + return p.read_text(encoding="utf-8") + return system_prompt + + +def normalize_tools(tools: Union[None, str, List[str]]) -> List[str]: + """Normalize tools to list of names. + str -> [str], list -> list, None -> [].""" + if tools is None: + return [] + if isinstance(tools, str): + return ( + [t.strip() for t in tools.split(",") if t.strip()] + if tools.strip() + else [] + ) + if isinstance(tools, list): + return [t if isinstance(t, str) else str(t) for t in tools] + return [] + + +async def ainput(prompt: str) -> str: + """Async input so event loop is not blocked.""" + loop = asyncio.get_running_loop() + return await loop.run_in_executor(None, lambda: input(prompt)) + + +def normalize_file_list(file: Union[None, str, List[str]]) -> List[str]: + """Normalize file to list of paths. None/str/list -> list of paths.""" + if file is None: + return [] + if isinstance(file, str): + return [p.strip() for p in file.split(",") if p.strip()] + return [str(p) for p in file] + + +async def run_agent_with_chat( # pylint: disable=too-many-branches + name: str, + system_prompt: Optional[str] = None, + tools: Union[None, str, List[str]] = None, + model: str = "qwen3-max", + task: Union[None, str] = None, + agent_type: str = "alias", + file: Union[None, str, List[str]] = None, + collection_name: Union[None, str] = None, +) -> None: + """ + Create agent (AliasAgentBase or QAAgent). If agent_type is 'qaagent', + create QAAgent with file/collection_name; else AliasAgentBase. + file: for QAAgent only (list of paths or comma-separated str). + task: if set, sent as first user message. + """ + if not os.environ.get("DASHSCOPE_API_KEY"): + print("DASHSCOPE_API_KEY not set, skip.") + return + + prompt_text = resolve_system_prompt(system_prompt) + tools_list = normalize_tools(tools) + agent_type = normalize_agent_type(agent_type) + file_list = normalize_file_list(file) if file is not None else None + + sandbox = None + worker_full_toolkit = None + try: + sandbox = AliasSandbox() + sandbox.__enter__() + except Exception as e: + print(f"Sandbox start failed: {e}") + print( + "Hint: docker run -d -p 6379:6379 " + "--name alias-redis redis:7-alpine", + ) + return + + try: + worker_full_toolkit = AliasToolkit(sandbox, add_all=True) + await add_tools(worker_full_toolkit) + + if agent_type == "qaagent": + agent = await QAAgent.create( + name=name, + model=model, + system_prompt=prompt_text or None, + tools=tools_list if tools_list else None, + worker_full_toolkit=worker_full_toolkit, + use_long_term_memory_service=False, + file=file_list, + collection_name=collection_name, + ) + else: + agent = await AliasAgentBase.create( + name=name, + model=model, + system_prompt=prompt_text or None, + tools=tools_list if tools_list else None, + worker_full_toolkit=worker_full_toolkit, + use_long_term_memory_service=False, + ) + + if task and task.strip(): + await agent( + Msg(name="user", content=task.strip(), role="user"), + ) + + while True: + user_input = await ainput( + "User (Enter `exit` or `quit` to exit): ", + ) + if not user_input or user_input.strip().lower() in ( + "exit", + "quit", + ): + print("Exiting.") + break + await agent( + Msg(name="user", content=user_input.strip(), role="user"), + ) + except (KeyboardInterrupt, asyncio.CancelledError): + print("\nInterrupted.") + except Exception as e: + print(f"Error: {e}") + traceback.print_exc() + finally: + if worker_full_toolkit is not None: + try: + await worker_full_toolkit.close_mcp_clients() + except Exception: + pass + if sandbox is not None: + try: + sandbox.__exit__(None, None, None) + except Exception: + pass + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Create an agent.", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__, + ) + parser.add_argument( + "--name", + "-n", + type=str, + required=True, + help="Agent name", + ) + parser.add_argument( + "--system_prompt", + "-s", + type=str, + default=None, + help="System prompt or path to file. None = agent default.", + ) + parser.add_argument( + "--tools", + "-t", + type=str, + default="", + help="Tool names, comma-separated or one. Empty = no extra tools.", + ) + parser.add_argument( + "--model", + "-m", + type=str, + default="qwen3-max", + help="Model name (default: qwen3-max)", + ) + parser.add_argument( + "--task", + type=str, + default="", + help="Initial question/task; if set, sent as first message.", + ) + parser.add_argument( + "--agent", + "-a", + type=str, + default="alias", + help="Agent type: 'qaagent' for QAAgent; else AliasAgentBase.", + ) + parser.add_argument( + "--file", + "-f", + type=str, + default=None, + nargs="*", + help="For QAAgent: RAG file path(s). Space- or comma-separated.", + ) + parser.add_argument( + "--collection_name", + type=str, + default=None, + help="For QAAgent: Qdrant collection name (default as_faq).", + ) + args = parser.parse_args() + + # Normalize --file: nargs='*' gives list or single element + file_arg = args.file + if ( + file_arg is not None + and isinstance(file_arg, list) + and len(file_arg) == 0 + ): + file_arg = None + if ( + file_arg is not None + and isinstance(file_arg, list) + and len(file_arg) == 1 + ): + file_arg = file_arg[0] if file_arg[0] else None + if file_arg is not None and isinstance(file_arg, list): + file_arg = [p for p in file_arg if p] + + if not sys.stdout.isatty(): + sys.stdout.reconfigure(line_buffering=True) + + _env_file, _created_env = _ensure_env() + try: + asyncio.run( + run_agent_with_chat( + name=args.name, + system_prompt=args.system_prompt, + tools=args.tools.strip() or None, + model=args.model, + task=args.task.strip() or None, + agent_type=args.agent, + file=file_arg, + collection_name=( + args.collection_name.strip() + if args.collection_name + else None + ), + ), + ) + finally: + if _created_env and _env_file and _env_file.exists(): + _env_file.unlink() + + +if __name__ == "__main__": + main() diff --git a/alias/src/alias/agent/agents/qa_agent_utils/create_rag_file.py b/alias/src/alias/agent/agents/qa_agent_utils/create_rag_file.py index af61dec3..a0412e25 100644 --- a/alias/src/alias/agent/agents/qa_agent_utils/create_rag_file.py +++ b/alias/src/alias/agent/agents/qa_agent_utils/create_rag_file.py @@ -87,6 +87,26 @@ def check_container_running(container_name: str) -> bool: return False +def collection_exists(collection_name: str) -> bool: + """ + Check if a Qdrant collection exists (Qdrant must be reachable). + Returns False if Qdrant is not reachable or collection does not exist. + """ + if not collection_name: + return False + try: + import urllib.request + + url = ( + f"http://{QDRANT_HOST}:{QDRANT_PORT}/collections/" + f"{collection_name}" + ) + with urllib.request.urlopen(url, timeout=2) as response: + return response.status == 200 + except Exception: + return False + + def start_qdrant_container() -> None: """Start Qdrant Docker container with specified storage location.""" if not check_docker_available():