diff --git a/phi/playground/operator.py b/phi/playground/operator.py new file mode 100644 index 0000000000..ff8ccf8c22 --- /dev/null +++ b/phi/playground/operator.py @@ -0,0 +1,44 @@ +from typing import List, Optional + +from phi.agent.agent import Agent, Tool, Toolkit, Function +from phi.utils.log import logger +from phi.agent.session import AgentSession + + +def format_tools(agent_tools): + formatted_tools = [] + if agent_tools is not None: + for tool in agent_tools: + if isinstance(tool, dict): + formatted_tools.append(tool) + elif isinstance(tool, Tool): + formatted_tools.append(tool.to_dict()) + elif isinstance(tool, Toolkit): + for f_name, f in tool.functions.items(): + formatted_tools.append(f.to_dict()) + elif isinstance(tool, Function): + formatted_tools.append(tool.to_dict()) + elif callable(tool): + func = Function.from_callable(tool) + formatted_tools.append(func.to_dict()) + else: + logger.warning(f"Unknown tool type: {type(tool)}") + return formatted_tools + + +def get_agent_by_id(agents: List[Agent], agent_id: str) -> Optional[Agent]: + for agent in agents: + if agent.agent_id == agent_id: + return agent + return None + + +def get_session_title(session: AgentSession) -> Optional[str]: + memory = session.memory + if memory is not None: + chat_history = memory.get("chat_history") + if chat_history is not None: + for history in chat_history: + if history.get("role") == "user": + return history.get("content") + return None diff --git a/phi/playground/routes.py b/phi/playground/routes.py index 2709069b1b..f1ecd737db 100644 --- a/phi/playground/routes.py +++ b/phi/playground/routes.py @@ -3,9 +3,10 @@ from fastapi import APIRouter, HTTPException, UploadFile from fastapi.responses import StreamingResponse, JSONResponse -from phi.agent.agent import Agent, RunResponse, Tool, Toolkit, Function +from phi.agent.agent import Agent, RunResponse from phi.utils.log import logger from phi.agent.session import AgentSession +from .operator import get_agent_by_id, get_session_title, format_tools from .schemas import ( AgentGetResponse, @@ -130,42 +131,3 @@ def agent_rename(body: AgentRenameRequest): return JSONResponse(content={"message": f"successfully renamed agent {agent.name}"}) return playground_routes - - -def format_tools(agent_tools): - formatted_tools = [] - if agent_tools is not None: - for tool in agent_tools: - if isinstance(tool, dict): - formatted_tools.append(tool) - elif isinstance(tool, Tool): - formatted_tools.append(tool.to_dict()) - elif isinstance(tool, Toolkit): - for f_name, f in tool.functions.items(): - formatted_tools.append(f.to_dict()) - elif isinstance(tool, Function): - formatted_tools.append(tool.to_dict()) - elif callable(tool): - func = Function.from_callable(tool) - formatted_tools.append(func.to_dict()) - else: - logger.warning(f"Unknown tool type: {type(tool)}") - return formatted_tools - - -def get_agent_by_id(agents: List[Agent], agent_id: str) -> Optional[Agent]: - for agent in agents: - if agent.agent_id == agent_id: - return agent - return None - - -def get_session_title(session: AgentSession) -> Optional[str]: - memory = session.memory - if memory is not None: - chat_history = memory.get("chat_history") - if chat_history is not None: - for history in chat_history: - if history.get("role") == "user": - return history.get("content") - return None