File size: 9,861 Bytes
5374a2d |
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 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 |
import re
from pydantic import Field, model_validator
from typing import Optional, List
from ..core.logging import logger
from ..core.module import BaseModule
from ..core.base_config import Parameter
from ..models.base_model import BaseLLM
from .action import Action, ActionInput, ActionOutput
from ..prompts.agent_generator import AGENT_GENERATION_ACTION
from ..prompts.tool_calling import AGENT_GENERATION_TOOLS_PROMPT
from ..utils.utils import normalize_text
class AgentGenerationInput(ActionInput):
"""
Input specification for the agent generation action.
"""
goal: str = Field(description="A detailed statement of the workflow's goal, explaining the objectives the entire workflow aims to achieve")
workflow: str = Field(description="An overview of the entire workflow, detailing all sub-tasks with their respective names, descriptions, inputs, and outputs")
task: str = Field(description="A detailed JSON representation of the sub-task requiring agent generation. It should include the task's name, description, inputs, and outputs.")
history: Optional[str] = Field(default=None, description="Optional field containing previously selected or generated agents.")
suggestion: Optional[str] = Field(default=None, description="Optional suggestions to refine the generated agents.")
existing_agents: Optional[str] = Field(default=None, description="Optional field containing the description of predefined agents, including each agent's name, role, and available actions.")
tools: Optional[str] = Field(default=None, description="Optional field containing the description of tools that agents can use, including each tool's name and functionality.")
class GeneratedAgent(BaseModule):
"""
Representation of a generated agent with validation capabilities.
"""
name: str
description: str
inputs: List[Parameter]
outputs: List[Parameter]
prompt: str
tool_names: Optional[List[str]] = None
@classmethod
def find_output_name(cls, text: str, outputs: List[str]):
def sim(t1: str, t2: str):
t1_words = normalize_text(t1).split()
t2_words = normalize_text(t2).split()
return len(set(t1_words)&set(t2_words))
similarities = [sim(text, output) for output in outputs]
max_sim = max(similarities)
return outputs[similarities.index(max_sim)]
@model_validator(mode="after")
@classmethod
def validate_prompt(cls, agent: 'GeneratedAgent'):
"""Validate and fix the agent's prompt template.
This validator ensures that:
1. All input parameters are properly referenced in the prompt
2. Input references use the correct format with braces
3. All output sections match the defined output parameters
If there are mismatches in the output sections, it attempts to
fix them by finding the most similar output name.
Args:
agent: The GeneratedAgent instance to validate.
Returns:
The validated and potentially modified GeneratedAgent.
Raises:
ValueError: If inputs are missing from the prompt or output sections don't match the defined outputs.
"""
# check whether all the inputs are present in the prompt
input_names = [inp.name for inp in agent.inputs]
prompt_has_inputs = [name in agent.prompt for name in input_names]
if not all(prompt_has_inputs):
missing_input_names = [name for name, has_input in zip(input_names, prompt_has_inputs) if not has_input]
raise ValueError(f'The prompt miss inputs: {missing_input_names}')
# check the format of the prompt to make sure it is wrapped in brackets.
pattern = r"### Instructions(.*?)### Output Format"
prompt = agent.prompt
def replace_with_braces(match):
instructions = match.group(1)
for name in input_names:
instructions = re.sub(fr'<input>{{*\b{re.escape(name)}\b}}*</input>', fr'<input>{{{name}}}</input>', instructions)
return "### Instructions" + instructions + "### Output Format"
modified_prompt = re.sub(pattern, replace_with_braces, prompt, flags=re.DOTALL)
agent.prompt = modified_prompt
# check whether all the outputs are present in the prompt
prompt = agent.prompt
pattern = r"### Output Format(.*)"
outputs_names = [out.name for out in agent.outputs]
def fix_output_names(match):
output_format = match.group(1)
matches = re.findall(r"## ([^\n#]+)", output_format, flags=re.DOTALL)
generated_outputs = [m.strip() for m in matches if m.strip() != "Thought"]
# check the number of generated outputs and agent outputs
if len(generated_outputs) != len(outputs_names):
raise ValueError(f"The number of outputs in the prompt is different from that defined in the `outputs` field of the agent. The outputs in the prompt are: {generated_outputs}, while the outputs from the agent's `outputs` field are: {outputs_names}")
# check whether the generated output names are the same as agent outputs
for generated_output in generated_outputs:
if generated_output not in outputs_names:
most_similar_output_name = cls.find_output_name(text=generated_output, outputs=outputs_names)
output_format = output_format.replace(generated_output, most_similar_output_name)
logger.warning(f"Couldn't find output name in prompt ('{generated_output}') in agent's outputs. Replace it with the most similar agent output: '{most_similar_output_name}'")
return "### Output Format" + output_format
modified_prompt = re.sub(pattern, fix_output_names, prompt, flags=re.DOTALL)
agent.prompt = modified_prompt
return agent
class AgentGenerationOutput(ActionOutput):
selected_agents: List[str] = Field(description="A list of selected agent's names")
generated_agents: List[GeneratedAgent] = Field(description="A list of generated agetns to address a sub-task")
class AgentGeneration(Action):
"""
Action for generating agent specifications for workflow tasks.
This action analyzes task requirements and generates appropriate agent
specifications, including their prompts, inputs, and outputs. It can either
select from existing agents or create new ones tailored to the task.
"""
def __init__(self, **kwargs):
name = kwargs.pop("name") if "name" in kwargs else AGENT_GENERATION_ACTION["name"]
description = kwargs.pop("description") if "description" in kwargs else AGENT_GENERATION_ACTION["description"]
prompt = kwargs.pop("prompt") if "prompt" in kwargs else AGENT_GENERATION_ACTION["prompt"]
# inputs_format = kwargs.pop("inputs_format") if "inputs_format" in kwargs else AgentGenerationInput
# outputs_format = kwargs.pop("outputs_format") if "outputs_format" in kwargs else AgentGenerationOutput
inputs_format = kwargs.pop("inputs_format", None) or AgentGenerationInput
outputs_format = kwargs.pop("outputs_format", None) or AgentGenerationOutput
tools = kwargs.pop("tools", None)
super().__init__(name=name, description=description, prompt=prompt, inputs_format=inputs_format, outputs_format=outputs_format, **kwargs)
self.tools = tools
def execute(self, llm: Optional[BaseLLM] = None, inputs: Optional[dict] = None, sys_msg: Optional[str]=None, return_prompt: bool = False, **kwargs) -> AgentGenerationOutput:
"""Execute the agent generation process.
This method uses the provided language model to generate agent specifications
based on the workflow context and task requirements.
Args:
llm: The language model to use for generation.
inputs: Input data containing workflow and task information.
sys_msg: Optional system message for the language model.
return_prompt: Whether to return both the generated agents and the prompt used.
**kwargs: Additional keyword arguments.
Returns:
If return_prompt is False (default): The generated agents output.
If return_prompt is True: A tuple of (generated agents, prompt used).
Raises:
ValueError: If the inputs are None or empty.
"""
if not inputs:
logger.error("AgentGeneration action received invalid `inputs`: None or empty.")
raise ValueError('The `inputs` to AgentGeneration action is None or empty.')
inputs_format: AgentGenerationInput = self.inputs_format
outputs_format: AgentGenerationOutput = self.outputs_format
prompt_params_names = inputs_format.get_attrs()
prompt_params_values = {param: inputs.get(param, "") for param in prompt_params_names}
if self.tools:
tool_description = [
{
tool.name: [
s["function"]["description"] for s in tool.get_tool_schemas()
],
}
for tool in self.tools
]
prompt_params_values["tools"] = AGENT_GENERATION_TOOLS_PROMPT.format(tools_description=tool_description)
prompt = self.prompt.format(**prompt_params_values)
agents = llm.generate(
prompt = prompt,
system_message = sys_msg,
parser=outputs_format,
parse_mode="json"
)
if return_prompt:
return agents, prompt
return agents
|