__init__.py 1.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950
  1. from enum import Enum
  2. from typing import List
  3. from sqlalchemy.orm import Session
  4. from sqlmodel import select
  5. from app.exceptions.exception import ServerError
  6. from app.models.action import Action
  7. from app.core.tools.base_tool import BaseTool
  8. from app.core.tools.external_function_tool import ExternalFunctionTool
  9. from app.core.tools.openapi_function_tool import OpenapiFunctionTool
  10. from app.core.tools.file_search_tool import FileSearchTool
  11. from app.core.tools.web_search import WebSearchTool
  12. # from app.core.tools.file_allcontent import FileContnetTool
  13. class AvailableTools(str, Enum):
  14. FILE_SEARCH = "file_search"
  15. WEB_SEARCH = "web_search"
  16. # FILE_CONTENT = "file_content"
  17. TOOLS = {
  18. AvailableTools.FILE_SEARCH: FileSearchTool,
  19. AvailableTools.WEB_SEARCH: WebSearchTool,
  20. # AvailableTools.FILE_CONTENT: FileContnetTool,
  21. }
  22. def find_tools(run, session: Session) -> List[BaseTool]:
  23. action_ids = [tool.get("id") for tool in run.tools if tool.get("type") == "action"]
  24. actions = (
  25. session.execute(select(Action).where(Action.id.in_(action_ids))).scalars().all()
  26. )
  27. action_map = {action.id: action for action in actions}
  28. tools = []
  29. for tool in run.tools:
  30. tool_name = tool["type"]
  31. if tool_name in TOOLS:
  32. tools.append(TOOLS[tool_name]())
  33. elif tool_name == "function":
  34. tools.append(ExternalFunctionTool(tool))
  35. elif tool_name == "action":
  36. action = action_map.get(tool.get("id"))
  37. tools.append(OpenapiFunctionTool(tool, run.extra_body, action))
  38. else:
  39. raise ServerError(f"Unknown tool type {tool}")
  40. return tools