6. 多模态文生图应用
6. 多模态文生图应用1. 概念介绍1.1 什么是文生图(Text-to-Image)?****核心原理2. 代码解析关键代码1. 工具层入口 (largemodel/utils/tools_manager.py)2. 模型接口层调度 (largemodel/utils/large_model_interface.py)代码解析3. 实践操作3.1 配置在线LLM3.2 启动并测试功能
在线文生图功能目前仅支持通义万相平台,无法获取通义万相平台API的用户如果要使用文生图功能,可以使用本地离线的fastsdcpu文生图方案。
1. 概念介绍
1.1 什么是文生图(Text-to-Image)?
文生图是一种通过文字描述 自动生成对应图像的人工智能技术。只需输入一段文字(例如“一只戴着墨镜的柴犬在沙滩上冲浪”),AI 模型就会根据语义理解生成符合描述的图片,无需任何绘画或设计基础。
核心原理
- 语言理解 :AI 先解析你的文字,识别关键词(如“柴犬”“墨镜”“沙滩”)。
- 图像生成 :基于海量图像数据训练,AI 将文字转化为视觉元素,并组合成合理的画面。
- 风格适配 :可指定风格(写实、卡通、水彩等),AI 会调整生成效果。
2. 代码解析
关键代码
1. 工具层入口 (largemodel/utils/tools_manager.py)
此函数定义了generate_image工具的完整执行流程。
x # From largemodel/utils/tools_manager.py class ToolsManager : # ... def generate_image ( self , args ): """ Generate image from text prompt and save to local file. 根据文本提示生成图像并保存到本地文件。 :param args: Arguments containing prompt and other parameters. :return: Dictionary with status and image paths. """ self . node . get_logger (). info ( f"Executing generate_image() tool with args: {args}" ) try : prompt = args . get ( "prompt" ) if isinstance ( args , dict ) else args if not prompt : self . node . get_logger (). error ( "Missing 'prompt' argument for generate_image." ) return { 'status' : 'failed' , 'error' : 'Missing prompt argument' } # 1. Call the text-to-image service to get the result. / 1. 调用文生图服务获取结果。 api_result = self . node . model_client . text_to_image ( prompt ) # 2. Check if the API call was successful and get the URL. / 2. 检查API调用是否成功,并获取URL。 if not isinstance ( api_result , dict ) or api_result . get ( 'status' ) ! = 'success' : error_msg = api_result . get ( 'error' , 'Failed to generate image from API' ) self . node . get_logger (). error ( f"Image generation API failed: {error_msg}" ) return { 'status' : 'failed' , 'error' : error_msg } image_urls = api_result . get ( "image_urls" , []) if not image_urls : self . node . get_logger (). error ( "API returned success but no image URLs found." ) return { 'status' : 'failed' , 'error' : 'No image URLs in API response' } # 3. Download and save the image. / 3. 下载并保存图片。 saved_paths = [] image_dir = os . path . join ( self . node . pkg_path , "resources_file" , "generated_images" ) os . makedirs ( image_dir , exist_ok = True ) for i , url in enumerate ( image_urls ): try : # Create a filename. / 创建一个文件名。 safe_prompt = re . sub ( r'[\\\/*?:"<>|]' , "" , prompt )[: 50 ] timestamp = datetime . now (). strftime ( "%Y%m%d_%H%M%S" ) filename = f"{safe_prompt}_{timestamp}_{i+1}.png" save_path = os . path . join ( image_dir , filename ) # Download image from URL. / 从URL下载图片。 urlretrieve ( url , save_path ) self . node . get_logger (). info ( f"Image successfully downloaded and saved to: {save_path}" ) saved_paths . append ( save_path ) # Immediately display the first successfully downloaded image. / 成功下载第一张图片后立即显示它。 if len ( saved_paths ) == 1 : try : generated_image = cv2 . imread ( save_path ) if generated_image is not None : cv2 . imshow ( "Generated Image" , generated_image ) cv2 . waitKey ( 1 ) except Exception as display_error : self . node . get_logger (). error ( f"Failed to display image {save_path}: {display_error}" ) except Exception as download_error : self . node . get_logger (). error ( f"Failed to download image from {url}: {download_error}" ) continue # If one image fails to download, continue with the next one. / 如果一张图片下载失败,继续尝试下一张。 if not saved_paths : return { 'status' : 'failed' , 'error' : 'Image generation succeeded, but all downloads failed.' } # 4. Return a success result containing the local path. / 4. 返回包含本地路径的成功结果。 return { 'status' : 'success' , 'image_urls' : image_urls , 'saved_paths' : saved_paths , 'message' : f"Successfully generated and saved {len(saved_paths)} image(s)." } except Exception as e : self . node . get_logger (). error ( f"Failed to execute generate_image tool: {e}" ) return { 'status' : 'failed' , 'error' : f'generate_image tool execution failed: {str(e)}' }
2. 模型接口层调度 (largemodel/utils/large_model_interface.py)
此文件中的函数负责根据配置与外部API进行实际的交互。
xxxxxxxxxx # From largemodel/utils/large_model_interface.py class model_interface : # ... def generate_image ( self , prompt , width = 1024 , height = 1024 , n = 1 ): """Text-to-image function interface, supporting multiple platforms. / 文生图功能接口,支持多平台。""" if self . llm_platform == 'tongyi' : return self . _tongyi_generate_image ( prompt , width , height , n ) elif self . llm_platform == 'ollama' : return self . _ollama_generate_image_fallback ( prompt , width , height , n ) else : return { 'status' : 'failed' , 'error' : f'The current platform ({self.llm_platform}) does not support text-to-image generation. Currently only Tongyi and Ollama platforms are supported.' } def _tongyi_generate_image ( self , prompt , width = 1024 , height = 1024 , n = 1 ): """Tongyi Qianwen text-to-image implementation. / 通义千问文生图实现。""" if not self . client : return "Tongyi client is not initialized." try : # Use dashscope SDK for text-to-image call / 使用dashscope SDK进行文生图调用 import dashscope dashscope . api_key = self . tongyi_api_key # Call text-to-image API / 调用文生图API response = dashscope . ImageSynthesis . call ( model = self . tongyi_media_model , prompt = prompt , n = n , size = f'{width}*{height}' ) if response . status_code == 200 : # Return the generated image URLs / 返回生成的图片URL image_urls = [ item [ 'url' ] for item in response . output [ 'results' ]] return { 'image_urls' : image_urls , 'status' : 'success' } else : return { 'error' : response . message , 'status' : 'failed' } except Exception as e : return { 'error' : str ( e ), 'status' : 'failed' } def text_to_image ( self , prompt , width = 1024 , height = 1024 , n = 1 ): """Independent text-to-image interface, specifically for generating images. / 独立的文生图接口,专门用于生成图像。""" # Directly call the generate_image method / 直接调用generate_image方法 return self . generate_image ( prompt , width , height , n )
代码解析
该功能的实现涉及两个主要层次:工具层负责业务流程,模型接口层负责与外部API交互。
-
模型接口层 (
large_model_interface.py):generate_image函数的作用是根据配置的self.llm_platform参数,选择对应的平台实现。- 如果平台是
'tongyi',它就把任务交给_tongyi_generate_image函数。对于其他平台,则返回错误。 _tongyi_generate_image函数的作用是封装API调用。它使用阿里云的dashscopeSDK,将prompt等参数发送到通义万相服务。请求成功后,它会从API的响应中解析出图片的URL,并将这些URL返回。
-
工具层 (
tools_manager.py):generate_image函数的作用是执行完整的业务流程。- 它首先调用模型接口层的
text_to_image方法,目的是获取一个包含图片URL的列表。 - 在拿到URL列表后,它会遍历这个列表,使用
urlretrieve函数下载每一张图片,并将它们保存到本地的resources_file/generated_images目录中。 - 最后,它将所有成功保存的本地文件路径作为最终结果返回给调用者。
总结来说,整个流程是:ToolsManager接收任务 -> model_interface根据配置选择并调用云服务API -> 云服务返回图片URL -> model_interface将URL返回给ToolsManager -> ToolsManager下载图片并保存 -> ToolsManager返回本地路径。
3. 实践操作
3.1 配置在线LLM
-
先从前面教程的任意一个平台中获取API Key之后获取API Key
-
然后需要更新配置文件中的key,打开模型接口配置文件
large_model_interface.yaml:
xxxxxxxxxx vim ~/yahboom_ws/src/largemodel/config/large_model_interface.yaml
3. 填入你的API Key : 找到对应的部分,将你刚刚复制的API Key粘贴进去。在线文生图目前只适配了通义万相
xxxxxxxxxx # large_model_interface.yaml ## 通义千问 qianwen_api_key : "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" # 粘贴你的Key tongyi_media_model : "flux-schnell" # 通义千问文生图模型
4. 打开主配置文件yahboom.yaml:
xxxxxxxxxx vim ~/yahboom_ws/src/largemodel/config/yahboom.yaml
5. 选择要使用的在线平台 : 修改llm_platform参数为你想要使用的平台名称
xxxxxxxxxx # yahboom.yaml model_service : ros__parameters : # ... llm_platform : 'tongyi' #在线文生图目前只适配了通义万相,所以选择qianwen
3.2 启动并测试功能
- 启动
largemodel主程序,并开启文本交互模式:
xxxxxxxxxx ros2 launch largemodel largemodel_control.launch.py text_chat_mode: = true
2. 发送文本指令 : 再次打开另一个终端,运行下面的指令,
xxxxxxxxxx ros2 run text_chat text_chat
然后可以开始输入你想提问的话。
-
测试 :
- 在终端中输入你的问题,然后按回车。例如:
生成一只在宇宙中翱翔的猪 - 观察终端输出,稍等片刻,就可以看到新生成的图片弹出。
- 在终端中输入你的问题,然后按回车。例如:
