agentic-ai-engineering/course/labs/L7-meta-agent/solution.py

179 lines
6.7 KiB
Python

"""
Lab 7.8: Meta-Agent -- SOLUTION
Generates a new agent persona from documentation.
"""
import json
import os
from datetime import datetime
AGENT_TEMPLATES = {
"system-prompt.md": """# {agent_name}
## Persona
{persona}
## Behavioral Rules
1. Always load mental model at session start
2. Log all tool calls with reasoning parameter
3. Escalate to human if confidence < 70%
4. Never run destructive operations
5. Update mental model after completing work
## Domain Expertise
{tools_description}
## Output Format
All outputs must include:
- Status indicator (+ working / - failed / ⚠ warning)
- Confidence level (HIGH / MEDIUM / LOW)
- Supporting evidence with file:line references
""",
"mental-model.yaml": """# {agent_name} Mental Model
# Auto-generated by meta-agent
# Last updated: {date}
expertise:
- topic: "initial setup"
notes: "Freshly created agent on {date}"
confidence: LOW
last_updated: "{date}"
observations: []
skills_to_develop: []
""",
"tools.py": """\"\"\"
{tool_descriptions}
\"\"\"
TOOLS = {tool_definitions}
def execute_tool(name: str, args: dict) -> str:
\"\"\"Execute a tool by name with given args.\"\"\"
handlers = {tool_handlers}
handler = handlers.get(name)
if handler:
return handler(args)
return f"Unknown tool: {{name}}"
"""
}
class MetaAgent:
def __init__(self, output_dir: str = "./generated-agents"):
self.output_dir = output_dir
def _infer_tools(self, description: str) -> list[dict]:
desc_lower = description.lower()
tools = []
if "security" in desc_lower or "vulnerability" in desc_lower:
tools.append({"name": "read_file", "desc": "Read file contents", "handler": "lambda a: f'Reading: {a[\"path\"]}'"})
tools.append({"name": "grep_search", "desc": "Search for patterns in files", "handler": "lambda a: f'Searching: {a[\"pattern\"]}'"})
tools.append({"name": "check_secrets", "desc": "Scan for exposed secrets", "handler": "lambda a: 'No secrets found'"})
tools.append({"name": "generate_report", "desc": "Generate security report", "handler": "lambda a: 'Report generated'"})
if "monitor" in desc_lower or "uptime" in desc_lower:
tools.append({"name": "check_endpoint", "desc": "Check HTTP endpoint health", "handler": "lambda a: 'Endpoint OK'"})
tools.append({"name": "check_certificate", "desc": "Check SSL certificate expiry", "handler": "lambda a: 'Cert OK'"})
tools.append({"name": "send_alert", "desc": "Send alert to channel", "handler": "lambda a: 'Alert sent'"})
if "data" in desc_lower or "analytics" in desc_lower or "analy" in desc_lower:
tools.append({"name": "query_database", "desc": "Execute SQL query", "handler": "lambda a: f'Query: {a[\"query\"]}'"})
tools.append({"name": "analyze_data", "desc": "Analyze dataset", "handler": "lambda a: 'Analysis complete'"})
if not tools:
tools.append({"name": "read_file", "desc": "Read files", "handler": "lambda a: f'Reading: {a[\"path\"]}'"})
tools.append({"name": "process_task", "desc": "Process task", "handler": "lambda a: f'Processing: {a}'"})
return tools
def _infer_persona(self, description: str) -> str:
desc_lower = description.lower()
if "security" in desc_lower:
return "Expert security engineer with 15 years of experience in application security, penetration testing, and vulnerability assessment."
if "monitor" in desc_lower:
return "SRE with expertise in system monitoring, alerting, and incident response."
if "data" in desc_lower or "analytics" in desc_lower:
return "Senior data engineer with expertise in SQL, data pipelines, and business intelligence."
if "support" in desc_lower or "customer" in desc_lower:
return "Customer support specialist with product knowledge and troubleshooting expertise."
return "General-purpose AI assistant with strong technical skills."
def build_agent(self, description: str) -> dict:
agent_name = description.strip().split("\n")[0].strip("- ").strip()[:30].replace(" ", "-").lower()
tools = self._infer_tools(description)
persona = self._infer_persona(description)
tool_names = [t["name"] for t in tools]
agent_dir = os.path.join(self.output_dir, agent_name)
os.makedirs(agent_dir, exist_ok=True)
files_created = []
# System prompt
sys_prompt = AGENT_TEMPLATES["system-prompt.md"].format(
agent_name=agent_name,
persona=persona,
tools_description="\n".join(f"- `{t['name']}`: {t['desc']}" for t in tools),
date=datetime.now().strftime("%Y-%m-%d")
)
with open(os.path.join(agent_dir, "system-prompt.md"), "w") as f:
f.write(sys_prompt)
files_created.append("system-prompt.md")
# Mental model
mental = AGENT_TEMPLATES["mental-model.yaml"].format(
agent_name=agent_name,
date=datetime.now().strftime("%Y-%m-%d")
)
with open(os.path.join(agent_dir, "mental-model.yaml"), "w") as f:
f.write(mental)
files_created.append("mental-model.yaml")
# Tools
tool_descriptions = f"Tools for {agent_name} agent."
tool_definitions = json.dumps(tools, indent=4)
tool_handlers = ",\n ".join('"' + t["name"] + '": ' + t["handler"] for t in tools)
tools_code = AGENT_TEMPLATES["tools.py"].format(
tool_descriptions=tool_descriptions,
tool_definitions=tool_definitions,
tool_handlers="{" + tool_handlers + "}"
)
with open(os.path.join(agent_dir, "tools.py"), "w") as f:
f.write(tools_code)
files_created.append("tools.py")
return {
"agent_name": agent_name,
"persona": persona,
"tools": tool_names,
"files_created": files_created,
"output_dir": agent_dir
}
if __name__ == "__main__":
meta = MetaAgent()
for description in [
"Security audit agent that reviews code for vulnerabilities",
"API uptime monitor with alerting",
"Data analytics agent for business intelligence",
"Customer support triage agent",
]:
print(f"\n{'='*55}")
print(f"Building: {description}")
print("=" * 55)
result = meta.build_agent(description)
print(f"Agent: {result['agent_name']}")
print(f"Persona: {result['persona']}")
print(f"Tools: {', '.join(result['tools'])}")
print(f"Files: {', '.join(result['files_created'])}")