Buckets:
| #!/usr/bin/env python3 | |
| import argparse | |
| import json | |
| import logging | |
| import os | |
| import sys | |
| import time | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Dict, List, Any, Optional | |
| sys.path.append(str(Path(__file__).parent / "src")) | |
| from src.agents.react_agent import ReActAgent, ReActResult | |
| from src.utils.llm_interface import LLMInterface | |
| from src.utils.environment import TravelEnvironment | |
| from src.core.memory import MemorySystem, EpisodicMemory, SemanticMemory | |
| from src.core.communication import MessageBroker | |
| def setup_logging(verbose: bool = False) -> logging.Logger: | |
| level = logging.DEBUG if verbose else logging.INFO | |
| logging.basicConfig( | |
| level=level, | |
| format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", | |
| handlers=[logging.StreamHandler(), logging.FileHandler("agent_pipeline.log")], | |
| ) | |
| return logging.getLogger(__name__) | |
| def create_travel_tools(environment: TravelEnvironment) -> Dict[str, callable]: | |
| def search_flights(destination: str, date: str) -> str: | |
| result = environment.search_flights(destination, date) | |
| if result["success"]: | |
| flights = result["flights"] | |
| if flights: | |
| flight = flights[0] | |
| return f"✅ Found flight {flight['flight_id']} to {destination} on {date} - {flight['airline']} - ${flight['price']}" | |
| else: | |
| return f"❌ No flights available to {destination} on {date}" | |
| else: | |
| return f"❌ {result['error']}" | |
| def book_flight(flight_id: str, destination: str, date: str) -> str: | |
| result = environment.book_flight(flight_id, destination, date) | |
| if result["success"]: | |
| return f"✅ Flight booked: {flight_id} to {destination} on {date}" | |
| else: | |
| return f"❌ Booking failed: {result['error']}" | |
| def search_hotels(destination: str, check_in_date: str, area: str = None) -> str: | |
| result = environment.search_hotels(destination, check_in_date, area) | |
| if result["success"]: | |
| hotels = result["hotels"] | |
| if hotels: | |
| hotel = hotels[0] | |
| return f"✅ Found hotel {hotel['name']} in {hotel['area']}, {destination} - ${hotel['price_per_night']}/night" | |
| else: | |
| return f"❌ No hotels available in {destination} for {check_in_date}" | |
| else: | |
| return f"❌ {result['error']}" | |
| def book_hotel( | |
| hotel_id: str, destination: str, check_in_date: str, area: str | |
| ) -> str: | |
| result = environment.book_hotel(hotel_id, destination, check_in_date, area) | |
| if result["success"]: | |
| return f"✅ Hotel booked: {result['booking']['hotel_name']} in {area}, {destination}" | |
| else: | |
| return f"❌ Booking failed: {result['error']}" | |
| def find_activities(destination: str, day: str) -> str: | |
| result = environment.search_activities(destination, day) | |
| if result["success"]: | |
| activities = result["activities"] | |
| if activities: | |
| activity_list = "\n".join( | |
| [f"- {act['name']}: {act['description']}" for act in activities[:3]] | |
| ) | |
| return f"🎯 Activities for {day} in {destination}:\n{activity_list}" | |
| else: | |
| return f"❌ No activities found for {day} in {destination}" | |
| else: | |
| return f"❌ {result['error']}" | |
| return { | |
| "search_flights": search_flights, | |
| "book_flight": book_flight, | |
| "search_hotels": search_hotels, | |
| "book_hotel": book_hotel, | |
| "find_activities": find_activities, | |
| } | |
| def run_react_demonstration( | |
| llm_interface: LLMInterface, | |
| tools: Dict[str, callable], | |
| environment: TravelEnvironment, | |
| logger: logging.Logger, | |
| ) -> Dict[str, Any]: | |
| logger.info("🚀 Starting ReAct Agent Demonstration") | |
| react_agent = ReActAgent( | |
| tools=tools, llm_interface=llm_interface, max_steps=8, agent_id="react_demo" | |
| ) | |
| test_cases = [ | |
| "I want to plan a trip to Paris for 2025-11-10. Book a flight and find a hotel.", | |
| "Search for flights to Tokyo on 2025-12-01 and book the first available flight.", | |
| "Find activities for Day 1 in New York and book a hotel in Manhattan.", | |
| ] | |
| results = [] | |
| for i, test_case in enumerate(test_cases, 1): | |
| logger.info(f"📋 ReAct Test Case {i}: {test_case}") | |
| environment.clear_bookings() | |
| start_time = time.time() | |
| result = react_agent.execute(test_case) | |
| execution_time = time.time() - start_time | |
| result_data = { | |
| "test_case": test_case, | |
| "success": result.success, | |
| "final_answer": result.final_answer, | |
| "steps_taken": len(result.steps), | |
| "execution_time": execution_time, | |
| "reasoning_quality": result.reasoning_quality, | |
| "tool_usage": result.tool_usage_stats, | |
| "error_messages": result.error_messages, | |
| "steps": [step.to_dict() for step in result.steps], | |
| } | |
| results.append(result_data) | |
| logger.info( | |
| f"✅ ReAct Test {i} completed: Success={result.success}, Steps={len(result.steps)}" | |
| ) | |
| total_tests = len(results) | |
| successful_tests = sum(1 for r in results if r["success"]) | |
| avg_execution_time = sum(r["execution_time"] for r in results) / total_tests | |
| avg_reasoning_quality = sum(r["reasoning_quality"] for r in results) / total_tests | |
| performance_summary = { | |
| "total_tests": total_tests, | |
| "successful_tests": successful_tests, | |
| "success_rate": successful_tests / total_tests, | |
| "avg_execution_time": avg_execution_time, | |
| "avg_reasoning_quality": avg_reasoning_quality, | |
| "agent_stats": react_agent.get_performance_stats(), | |
| } | |
| return { | |
| "architecture": "ReAct", | |
| "results": results, | |
| "performance_summary": performance_summary, | |
| "environment_stats": environment.get_environment_stats(), | |
| } | |
| def run_comprehensive_demonstration( | |
| config: Dict[str, Any], output_dir: Path, logger: logging.Logger | |
| ) -> Dict[str, Any]: | |
| logger.info("🎭 Starting Comprehensive Language Agent Architecture Demonstration") | |
| logger.info("🔧 Initializing components...") | |
| llm_config = config.get("llm", {}) | |
| llm_interface = LLMInterface( | |
| primary_provider=llm_config.get("primary_provider", "mock"), | |
| fallback_providers=llm_config.get("fallback_providers", ["local", "mock"]), | |
| config=llm_config, | |
| ) | |
| environment = TravelEnvironment() | |
| tools = create_travel_tools(environment) | |
| memory_system = MemorySystem( | |
| db_path=":memory:" | |
| ) | |
| episodic_memory = EpisodicMemory(memory_system) | |
| semantic_memory = SemanticMemory(memory_system) | |
| message_broker = MessageBroker() | |
| logger.info("✅ All components initialized successfully") | |
| all_results = {} | |
| logger.info("=" * 60) | |
| react_results = run_react_demonstration(llm_interface, tools, environment, logger) | |
| all_results["react"] = react_results | |
| final_report = { | |
| "timestamp": datetime.now().isoformat(), | |
| "config": config, | |
| "environment_stats": environment.get_environment_stats(), | |
| "llm_provider_info": llm_interface.get_provider_info(), | |
| "results": all_results, | |
| "summary": { | |
| "total_architectures_tested": len(all_results), | |
| "overall_success_rate": ( | |
| sum( | |
| r["performance_summary"]["success_rate"] | |
| for r in all_results.values() | |
| ) | |
| / len(all_results) | |
| if all_results | |
| else 0 | |
| ), | |
| }, | |
| } | |
| return final_report | |
| def save_results(results: Dict[str, Any], output_dir: Path, logger: logging.Logger): | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| results_file = output_dir / "demonstration_results.json" | |
| with open(results_file, "w", encoding="utf-8") as f: | |
| json.dump(results, f, indent=2, ensure_ascii=False) | |
| logger.info(f"📁 Results saved to {results_file}") | |
| for arch_name, arch_results in results["results"].items(): | |
| arch_file = output_dir / f"{arch_name}_results.json" | |
| with open(arch_file, "w", encoding="utf-8") as f: | |
| json.dump(arch_results, f, indent=2, ensure_ascii=False) | |
| logger.info(f"📁 {arch_name} results saved to {arch_file}") | |
| summary_file = output_dir / "performance_summary.json" | |
| summary_data = { | |
| "timestamp": results["timestamp"], | |
| "summary": results["summary"], | |
| "environment_stats": results["environment_stats"], | |
| "llm_provider_info": results["llm_provider_info"], | |
| } | |
| with open(summary_file, "w", encoding="utf-8") as f: | |
| json.dump(summary_data, f, indent=2, ensure_ascii=False) | |
| logger.info(f"📁 Performance summary saved to {summary_file}") | |
| def main(): | |
| parser = argparse.ArgumentParser( | |
| description="Language Agent Architecture Pipeline Demonstration", | |
| formatter_class=argparse.RawDescriptionHelpFormatter, | |
| epilog= | |
| ) | |
| parser.add_argument( | |
| "--config", | |
| type=str, | |
| default="config.json", | |
| help="Configuration file path (default: config.json)", | |
| ) | |
| parser.add_argument( | |
| "--output", | |
| type=str, | |
| default="results", | |
| help="Output directory for results (default: results)", | |
| ) | |
| parser.add_argument("--verbose", action="store_true", help="Enable verbose logging") | |
| args = parser.parse_args() | |
| logger = setup_logging(args.verbose) | |
| try: | |
| logger.info("🎯 Language Agent Architecture Pipeline Starting") | |
| logger.info(f"📁 Output directory: {args.output}") | |
| logger.info(f"⚙️ Configuration file: {args.config}") | |
| config_path = Path(args.config) | |
| if config_path.exists(): | |
| with open(config_path, "r", encoding="utf-8") as f: | |
| if config_path.suffix.lower() == ".json": | |
| config = json.load(f) | |
| else: | |
| import yaml | |
| config = yaml.safe_load(f) | |
| else: | |
| logger.warning( | |
| f"Configuration file {args.config} not found, using defaults" | |
| ) | |
| config = { | |
| "llm": { | |
| "primary_provider": "mock", | |
| "fallback_providers": ["local", "mock"], | |
| }, | |
| "agents": {"max_steps": 10, "temperature": 0.3}, | |
| } | |
| output_dir = Path(args.output) | |
| start_time = time.time() | |
| results = run_comprehensive_demonstration(config, output_dir, logger) | |
| total_time = time.time() - start_time | |
| save_results(results, output_dir, logger) | |
| logger.info("=" * 60) | |
| logger.info("🎉 DEMONSTRATION COMPLETED SUCCESSFULLY") | |
| logger.info(f"⏱️ Total execution time: {total_time:.2f} seconds") | |
| logger.info( | |
| f"📊 Architectures tested: {results['summary']['total_architectures_tested']}" | |
| ) | |
| logger.info( | |
| f"✅ Overall success rate: {results['summary']['overall_success_rate']:.1%}" | |
| ) | |
| logger.info(f"📁 Results saved to: {output_dir}") | |
| logger.info("=" * 60) | |
| except KeyboardInterrupt: | |
| logger.info("🛑 Demonstration interrupted by user") | |
| sys.exit(1) | |
| except Exception as e: | |
| logger.error(f"❌ Demonstration failed: {e}") | |
| if args.verbose: | |
| import traceback | |
| logger.error(traceback.format_exc()) | |
| sys.exit(1) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 11.8 kB
- Xet hash:
- 8e687c7715f444e955f89100a016c309278a6752b3b93cc49ba69df116e5ee31
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.