kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

style: auto-format with ruff [skip ci]

Browse Source
This commit is contained in:
GitHub Actions 2025-12-30 12:12:04 +00:00
parent c8cb4edf5b
commit 834202fc20
2 changed files with 37 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/agents/common/tools.py
View File

@ -124,6 +124,7 @@ class KnowledgeRetrieverModel(BaseModel):
class CommonKnowledgeRetriever(KnowledgeRetrieverModel): class CommonKnowledgeRetriever(KnowledgeRetrieverModel):
"""Common knowledge retriever model.""" """Common knowledge retriever model."""
file_name: str = Field(description="限定文件名称，当操作类型为 'search' 时，可以指定文件名称，支持模糊匹配") file_name: str = Field(description="限定文件名称，当操作类型为 'search' 时，可以指定文件名称，支持模糊匹配")

22
test/test_milvus_filter.py
View File

@ -1,11 +1,11 @@
import asyncio import asyncio
import os import os
import shutil from unittest.mock import patch
from unittest.mock import MagicMock, patch
from src.knowledge import knowledge_base from src.knowledge import knowledge_base
from src.utils import logger from src.utils import logger
# Mock Embedding Model # Mock Embedding Model
class MockEmbeddingModel: class MockEmbeddingModel:
async def abatch_encode(self, texts, batch_size=None): async def abatch_encode(self, texts, batch_size=None):
@ -15,13 +15,15 @@ class MockEmbeddingModel:
def batch_encode(self, texts, batch_size=None): def batch_encode(self, texts, batch_size=None):
return [[0.1, 0.2, 0.3, 0.4] for _ in texts] return [[0.1, 0.2, 0.3, 0.4] for _ in texts]
# Test function # Test function
async def test_milvus_filter(): async def test_milvus_filter():
logger.info("Starting Milvus Filter Test") logger.info("Starting Milvus Filter Test")
# Check if Milvus is available (pymilvus installed and connection works) # Check if Milvus is available (pymilvus installed and connection works)
try: try:
from pymilvus import connections, utility from pymilvus import connections
# Assuming Milvus is running at default location # Assuming Milvus is running at default location
connections.connect(alias="default", uri=os.getenv("MILVUS_URI", "http://localhost:19530")) connections.connect(alias="default", uri=os.getenv("MILVUS_URI", "http://localhost:19530"))
logger.info("Connected to Milvus") logger.info("Connected to Milvus")
@ -35,7 +37,6 @@ async def test_milvus_filter():
# Patch embedding model # Patch embedding model
with patch("src.models.embed.select_embedding_model", return_value=MockEmbeddingModel()): with patch("src.models.embed.select_embedding_model", return_value=MockEmbeddingModel()):
try: try:
# Cleanup if exists # Cleanup if exists
if db_id in knowledge_base.global_databases_meta: if db_id in knowledge_base.global_databases_meta:
@ -48,11 +49,13 @@ async def test_milvus_filter():
database_name="Test Milvus Filter", database_name="Test Milvus Filter",
description="Test DB", description="Test DB",
kb_type="milvus", kb_type="milvus",
embed_info={"name": "mock-embedding", "dimension": 4, "model_id": "mock"} embed_info={"name": "mock-embedding", "dimension": 4, "model_id": "mock"},
) )
# Get actual db_id # Get actual db_id
target_db = next((db for db in knowledge_base.get_databases()["databases"] if db["name"] == "Test Milvus Filter"), None) target_db = next(
(db for db in knowledge_base.get_databases()["databases"] if db["name"] == "Test Milvus Filter"), None
)
if not target_db: if not target_db:
logger.error("Failed to create DB") logger.error("Failed to create DB")
return return
@ -81,7 +84,7 @@ async def test_milvus_filter():
logger.info(f"No filter results: {len(results)}") logger.info(f"No filter results: {len(results)}")
# Verify we have chunks from both files # Verify we have chunks from both files
sources = [r['metadata']['source'] for r in results] sources = [r["metadata"]["source"] for r in results]
logger.info(f"Sources: {sources}") logger.info(f"Sources: {sources}")
# Query with filter A (Partial Match) # Query with filter A (Partial Match)
@ -93,7 +96,7 @@ async def test_milvus_filter():
logger.error("FAIL: Filter A returned 0 results") logger.error("FAIL: Filter A returned 0 results")
for r in results_a: for r in results_a:
source = r['metadata']['source'] source = r["metadata"]["source"]
logger.info(f" - {source}") logger.info(f" - {source}")
if "test_file_A.txt" not in source: if "test_file_A.txt" not in source:
logger.error(f"FAIL: Expected test_file_A.txt, got {source}") logger.error(f"FAIL: Expected test_file_A.txt, got {source}")
@ -107,7 +110,7 @@ async def test_milvus_filter():
logger.error("FAIL: Wildcard filter returned 0 results") logger.error("FAIL: Wildcard filter returned 0 results")
for r in results_b: for r in results_b:
source = r['metadata']['source'] source = r["metadata"]["source"]
logger.info(f" - {source}") logger.info(f" - {source}")
if "test_file_B.txt" not in source: if "test_file_B.txt" not in source:
logger.error(f"FAIL: Expected test_file_B.txt, got {source}") logger.error(f"FAIL: Expected test_file_B.txt, got {source}")
@ -133,5 +136,6 @@ async def test_milvus_filter():
if os.path.exists(file2): if os.path.exists(file2):
os.remove(file2) os.remove(file2)
if __name__ == "__main__": if __name__ == "__main__":
asyncio.run(test_milvus_filter()) asyncio.run(test_milvus_filter())