|
|
|
|
|
"""tools.py |
|
|
|
|
|
Automatically generated by Colab. |
|
|
|
|
|
Original file is located at |
|
|
https://colab.research.google.com/drive/1Srm2oS1q5iYwlo4PoMx6TBc2uNoSVao3 |
|
|
""" |
|
|
|
|
|
from typing import Dict, Any, List |
|
|
from langchain.tools import Tool |
|
|
|
|
|
"""**It's purely deterministic Python, no LLM inside the tool, so the reasoning stays transparent.**""" |
|
|
|
|
|
def modernize_stack(legacy_stack: Dict[str, Any]) -> List[str]: |
|
|
"""Return concrete modernization actions based on a given legacy stack.""" |
|
|
actions: List[str] = [] |
|
|
|
|
|
scheduler = str(legacy_stack.get("scheduler", "")).lower() |
|
|
language = str(legacy_stack.get("language", "")).lower() |
|
|
architecture = str(legacy_stack.get("architecture", "")).lower() |
|
|
dependencies = [d.lower() for d in legacy_stack.get("dependencies", [])] |
|
|
|
|
|
if "autosys" in scheduler: |
|
|
actions.append("Replace Autosys with Apache Airflow or AWS Step Functions.") |
|
|
|
|
|
if "java 8" in language: |
|
|
actions.append("Upgrade Java 8 → Java 17 and containerize the application.") |
|
|
|
|
|
if "monolith" in architecture: |
|
|
actions.append("Refactor monolith into microservices and add an API Gateway.") |
|
|
|
|
|
if any("ftp" in dep for dep in dependencies): |
|
|
actions.append("Replace FTP with S3 and event-driven ingestion via Lambda.") |
|
|
|
|
|
actions.append("Adopt Terraform for Infrastructure-as-Code.") |
|
|
|
|
|
return actions |
|
|
|
|
|
"""**This tool takes both the legacy_stack and an outage_scenario string to generate resilience test ideas.**""" |
|
|
|
|
|
def generate_resilience_plan(legacy_stack: Dict[str, Any], outage_scenario: str) -> List[str]: |
|
|
"""Return a resilience test plan for a given outage scenario.""" |
|
|
plan: List[str] = [] |
|
|
|
|
|
if "oracle" in outage_scenario.lower(): |
|
|
plan.append("Test RDS Multi-AZ failover for Oracle DB.") |
|
|
plan.append("Implement circuit breaker around DB calls.") |
|
|
|
|
|
if "kafka" in outage_scenario.lower(): |
|
|
plan.append("Simulate Kafka broker outage and validate consumer recovery.") |
|
|
|
|
|
plan.append("Perform chaos engineering tests for network partition.") |
|
|
plan.append("Run DR simulation and validate RTO/RPO.") |
|
|
|
|
|
return plan |
|
|
|
|
|
"""**This step is critical, it's how the agent knows these Python functions exist and what they do.** |
|
|
|
|
|
- Deterministic: No AI hallucinations in the tools, pure logic. |
|
|
|
|
|
- Composable: We can add more tools later (cost analysis, migration path scorer). |
|
|
|
|
|
- Agent-Controlled: The agent can decide when to call each tool based on the request. |
|
|
""" |
|
|
|
|
|
def get_tools() -> List[Tool]: |
|
|
"""Return a list of LangChain Tools for the agent.""" |
|
|
return [ |
|
|
Tool( |
|
|
name="modernize_stack", |
|
|
func=lambda x: "\n".join(modernize_stack(x)), |
|
|
description="Recommend modernization actions for a given legacy stack." |
|
|
), |
|
|
Tool( |
|
|
name="generate_resilience_plan", |
|
|
func=lambda x: "\n".join(generate_resilience_plan( |
|
|
x.get("legacy_stack", {}), |
|
|
x.get("outage_scenario", "") |
|
|
)), |
|
|
description="Generate a resilience test plan for the given legacy stack and outage scenario." |
|
|
) |
|
|
] |
