Spaces:
Runtime error
Runtime error
File size: 3,349 Bytes
0af0a55 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 |
from typing import Dict, Any, List
from loguru import logger
from utils.llm_orchestrator import LLMOrchestrator
class PlanningAgent:
def __init__(self, llm_api_key: str):
"""Initialize the Planning Agent."""
logger.info("Initializing PlanningAgent")
self.llm_orchestrator = LLMOrchestrator(llm_api_key)
self.capabilities = [
"task_planning",
"goal_decomposition",
"plan_refinement",
"task_prioritization"
]
self.setup_logger()
def setup_logger(self):
"""Configure logging for the agent."""
logger.add("logs/planning_agent.log", rotation="500 MB")
async def generate_plan(
self, goal: str, available_agents: List[str]) -> Dict[str, Any]:
"""Generate a task plan based on a high-level goal."""
logger.info(f"Generating plan for goal: {goal}")
try:
prompt = f"""
You are an expert planner. Generate a detailed task plan to achieve the following goal:
Goal: {goal}
Available agents: {', '.join(available_agents)}
Think step-by-step and explain your reasoning for each step.
The plan should be a list of steps, each with:
- A clear description of the task.
- The agent best suited to execute the task.
- Any necessary input or parameters for the task.
Example:
1. Task: Summarize the latest news on topic X.
Agent: web_browsing_agent
Input: topic=X
Reasoning: To get the latest news, we need to use the web_browsing_agent to search for news on topic X.
2. Task: Analyze the sentiment of the news summary.
Agent: data_analysis_agent
Input: summary from step 1
Reasoning: To analyze the sentiment, we can use the data_analysis_agent to process the summary from the previous step.
"""
plan_str = await self.llm_orchestrator.generate_completion(prompt)
plan = self.parse_plan(plan_str)
logger.info(f"Plan generated successfully: {plan}")
return {
"status": "success",
"plan": plan
}
except Exception as e:
logger.error(f"Error generating plan: {str(e)}")
return {
"status": "error",
"message": str(e)
}
def parse_plan(self, plan_str: str) -> List[Dict[str, Any]]:
"""Parse the plan generated by the LLM into a structured format."""
plan = []
steps = plan_str.strip().split("\n")
current_step = {}
for line in steps:
if line.startswith(tuple(f"{i}." for i in range(1, 10))):
if current_step:
plan.append(current_step)
current_step = {"task": line.split("Task: ")[1]}
elif "Agent: " in line:
current_step["agent"] = line.split("Agent: ")[1]
elif "Input: " in line:
current_step["input"] = line.split("Input: ")[1]
elif "Reasoning: " in line:
current_step["reasoning"] = line.split("Reasoning: ")[1]
if current_step:
plan.append(current_step)
return plan
|