使用Strands Agents、Meta Llama 4和Amazon Bedrock创建多智能体解决方案

def new_llama4_coordinator_agent() -> Agent:
    """
    Factory constructor: creates a NEW agent instance with a fresh conversation history.
    Use this per video request for clean isolation.
    """
    return Agent(
            system_prompt="""You are a video processing coordinator. Your job is to process videos step by step.
##When asked to process a video:
1. Extract frames from S3 video using run_frame_extraction
2. Use the frame location from step 1 to run_visual_analysis
3. WAIT for visual analysis to complete sending the json to s3
4. Use the retrieve_json agent to extract the json from step 3
5. Use the text result of retrieve_json_from_s3 by passing it to run_temporal_reasoning
6. Pass the result from temporal reasoning to run_summary_generation
7. Upload analysis generated in run_summary_generation and return s3 location
##IMPORTANT:
- Call ONE tool at a time and wait for the result
- Use the EXACT result from the previous step as input
- Do NOT call multiple tools simultaneously
- Do NOT return raw JSON or function call syntax
""",
            model=bedrock_model,
            tools=[
                run_frame_extraction,
                run_visual_analysis,
                run_temporal_reasoning,
                run_summary_generation,
                upload_analysis_results,
                retrieve_json_from_s3,
            ],
    )

@tool
def upload_local_json_to_s3(s3_video_path: str, local_filename: str = "visual_analysis_results.json") -> str:
    """Upload local JSON file to S3 bucket in video folder"""
    try:
        s3_parts = [part for part in s3_video_path.replace('s3://', '').split('/') if part]
        bucket = s3_parts[0]
        video_folder = s3_parts[-1]
        
        if '_' in video_folder:
            base_video_name = video_folder.split('_')[0]
        else:
            base_video_name = video_folder
        random_num = randint(1000, 9999)
        
        s3_key = f"videos/{base_video_name}/{random_num}_{local_filename}"
        
        s3_client = boto3.client('s3')
        s3_client.upload_file(local_filename, bucket, s3_key)
        
        return f"s3://{bucket}/{s3_key}"
    except Exception as e:
        return f"Error uploading file: {str(e)}"


s_visual_analysis_agent = Agent(
    system_prompt="""You are an image analysis agent that processes frames from S3 buckets. Your workflow:
1. Use the available tools to analyze images
2. Use the video path folder to place the analysis results IMPORTANT:
- Do NOT generate, write, or return any code
- Focus on describing what you see in the images
- Images are automatically resized if too large
- Put numbered labels in front of each image description (e.g., "1. ", "2. ", etc.)
- Always save analysis results locally first, then upload to S3 Return Format:
The uri from the upload_local_json_to_s3 tool""",
    model=bedrock_model,
    callback_handler=None,    tools=[list_s3_frames, analyze_image, analyze_all_frames, analyze_frames_batch, upload_local_json_to_s3],
)

@tool
def process_s3_analysis_json(s3_uri: str) -> str:
    """Retrieve JSON from S3 and extract only the analysis text"""
    try:
        # Parse S3 URI and download JSON
        s3_parts = s3_uri.replace('s3://', '').split('/', 1)
        bucket = s3_parts[0]
        key = s3_parts[1]
        
        s3_client = boto3.client('s3')
        response = s3_client.get_object(Bucket=bucket, Key=key)
        json_content = response['Body'].read().decode('utf-8')
        
        # Parse and extract text
        data = json.loads(json_content)
        
        # Handle both formats
        if 'analyses' in data:
            analyses = data['analyses']
        elif 'sessions' in data:
            analyses = [session['data'] for session in data['sessions'] if 'data' in session]
        else:
            return "Error: No 'analyses' or 'sessions' field found"
        
        # Extract text only
        text_only = []
        for analysis in analyses:
            if 'analysis' in analysis:
                text = analysis['analysis']
                if not text.startswith("Failed:"):
                    text_only.append(text)
        
        # Clean up local file
        local_file = "visual_analysis_results.json"
        if os.path.exists(local_file):
            os.remove(local_file)
        
        return "\n".join(text_only)
    except Exception as e:
        return f"Error processing {s3_uri}: {str(e)}"

bedrock_model = BedrockModel(
    model_id='us.meta.llama4-maverick-17b-instruct-v1:0',
    region_name=region,
    streaming=False,
    temperature=0
)

retrieve_json_agent = Agent(
    system_prompt="Call process_s3_analysis_json with the S3 URI. Your response must be the exact text output from the tool, nothing else.",
    model=bedrock_model,
    callback_handler=None,    tools=[process_s3_analysis_json],
)

git clone  https://github.com/aws-samples/Meta-Llama-on-AWS.git
cd agents/strands/Bedrock/multi-agent-video-processing/

pip install -r requirements.txt

Llama Video Analysis Log
Flik is shown determined in front of a tree.
He interacts with other insects.
Flik gathers items and constructs a device.
He presents the invention to a group of insects.
The group reacts withskepticism.
Flik is chased by a group of birds. Key visual elements:
The key visual elements include Flik’s determined expression, his interaction with other insects, the items he gathers, the complex device he constructs, the group’s skeptical reaction, and the chaotic scene of Flik being chased by birds.
Overall Narrative:
The narrative follows Flik’s journey as he prepares and presents an invention, faces rejection, and experiences a dramatic consequence. The story is character-driven, showcasing Flik’s actions and their outcomes, and builds up to a climactic event.

def upload_to_sagemaker_bucket(local_video_path, base_folder="videos/"):
    sagemaker = boto3.client('sagemaker')
    s3 = boto3.client('s3')
    # Get default SageMaker bucket
    account_id = boto3.client('sts').get_caller_identity()['Account']
    region = boto3.Session().region_name
    bucket_name = f"sagemaker-{region}-{account_id}"
    # Get filename and create subfolder name
    filename = os.path.basename(local_video_path)
    filename_without_ext = os.path.splitext(filename)[0]
    # Create the full S3 path: videos/filename_without_ext/filename
    s3_key = os.path.join(base_folder, filename_without_ext, filename)
    # Upload file
    s3.upload_file(local_video_path, bucket_name, s3_key)
    
    s3_uri = f"s3://{bucket_name}/{s3_key}"
    print(f"Uploaded to {s3_uri}")
        
    s3_folder_path = os.path.join(base_folder, filename_without_ext)
    s3_folder_uri = f"s3://{bucket_name}/{s3_folder_path}"
    
    return s3_folder_uri # Example usage: Provide your local video path here