在这篇教程中,我将带你了解如何无缝集成 AutoGen 和 Semantic Kernel 与 Google 的 Gemini Flash 模型。我们首先设置 GeminiWrapper 和 SemanticKernelGeminiPlugin 类,将 Gemini 的生成能力与 AutoGen 的多智能体编排功能连接起来。接下来,我们配置专业智能体,包括代码审查员和创意分析师,展示如何利用 AutoGen 的 ConversableAgent API 结合 Semantic Kernel 的装饰器函数来进行文本分析、摘要生成、代码审查和创意问题解决。

通过结合 AutoGen 强大的智能体框架和 Semantic Kernel 基于函数的方法,我们创建了一个高级 AI 助手,能够适应各种任务并提供结构化、可操作的见解。

环境设置和依赖安装

首先,我们需要安装核心依赖项:

1
!pip install pyautogen semantic-kernel google-generativeai python-dotenv

然后导入必要的模块:

1
2
3
4
5
6
7
8
import os
import asyncio
from typing import Dict, Any, List
import autogen
import google.generativeai as genai
from semantic_kernel import Kernel
from semantic_kernel.functions import KernelArguments
from semantic_kernel.functions.kernel_function_decorator import kernel_function

配置 Gemini API

接下来配置 Gemini API 密钥和模型设置:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
GEMINI_API_KEY = "your-api-key-here" 
genai.configure(api_key=GEMINI_API_KEY)

config_list = [
   {
       "model": "gemini-1.5-flash",
       "api_key": GEMINI_API_KEY,
       "api_type": "google",
       "api_base": "https://generativelanguage.googleapis.com/v1beta",
   }
]

我们定义 GEMINI_API_KEY 并立即配置 genai 客户端,确保所有后续的 Gemini 调用都经过身份验证。然后构建包含 Gemini Flash 模型设置的 config_list,包括模型名称、API 密钥、端点类型和基础 URL,这些将传递给我们的智能体进行 LLM 交互。

创建 Gemini 包装器类

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
class GeminiWrapper:
    """Gemini API 包装器,用于与 AutoGen 协同工作"""
    
    def __init__(self, model_name="gemini-1.5-flash"):
        self.model = genai.GenerativeModel(model_name)
    
    def generate_response(self, prompt: str, temperature: float = 0.7) -> str:
        """使用 Gemini 生成响应"""
        try:
            response = self.model.generate_content(
                prompt,
                generation_config=genai.types.GenerationConfig(
                    temperature=temperature,
                    max_output_tokens=2048,
                )
            )
            return response.text
        except Exception as e:
            return f"Gemini API 错误: {str(e)}"

这个 GeminiWrapper 类封装了所有 Gemini Flash 交互,我们初始化一个 GenerativeModel 并公开一个简单的 generate_response 方法。该方法将提示和温度参数传递给 Gemini 的 generate_content API(限制在 2048 个令牌),并返回原始文本或格式化的错误信息。

构建 Semantic Kernel 插件

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class SemanticKernelGeminiPlugin:
    """使用 Gemini Flash 进行高级 AI 操作的 Semantic Kernel 插件"""
    
    def __init__(self):
        self.kernel = Kernel()
        self.gemini = GeminiWrapper()
    
    @kernel_function(
        name="analyze_text", 
        description="分析文本的情感和关键见解"
    )
    def analyze_text(self, text: str) -> str:
        """使用 Gemini Flash 分析文本"""
        prompt = f"""
        全面分析以下文本:
        
        文本: {text}
        
        请按以下格式提供分析:
        - 情感:[积极/消极/中性,包含置信度]
        - 关键主题:[主要话题和概念]
        - 见解:[重要观察和模式]
        - 建议:[可操作的后续步骤]
        - 语调:[正式/非正式/技术/情感]
        """
        
        return self.gemini.generate_response(prompt, temperature=0.3)
    
    @kernel_function(
        name="generate_summary", 
        description="生成综合摘要"
    )
    def generate_summary(self, content: str) -> str:
        """使用 Gemini 的高级能力生成摘要"""
        prompt = f"""
        为以下内容创建综合摘要:
        
        内容: {content}
        
        请提供:
        1. 执行摘要(2-3句话)
        2. 要点(项目符号格式)
        3. 重要细节
        4. 结论/影响
        """
        
        return self.gemini.generate_response(prompt, temperature=0.4)
    
    @kernel_function(
        name="code_analysis", 
        description="分析代码质量并提供建议"
    )
    def code_analysis(self, code: str) -> str:
        """使用 Gemini 的代码理解能力分析代码"""
        prompt = f"""
        全面分析这段代码:
        
        ```
        {code}
        ```
        
        请提供分析,涵盖:
        - 代码质量:[可读性、结构、最佳实践]
        - 性能:[效率、优化机会]
        - 安全性:[潜在漏洞、安全最佳实践]
        - 可维护性:[文档、模块化、可扩展性]
        - 建议:[具体改进和示例]
        """
        
        return self.gemini.generate_response(prompt, temperature=0.2)
    
    @kernel_function(
        name="creative_solution", 
        description="为问题生成创意解决方案"
    )
    def creative_solution(self, problem: str) -> str:
        """使用 Gemini 的创意能力生成创意解决方案"""
        prompt = f"""
        问题: {problem}
        
        生成创意解决方案:
        1. 传统方法(2-3个标准解决方案)
        2. 创新想法(3-4个创意替代方案)
        3. 混合解决方案(结合不同方法)
        4. 实施策略(实用步骤)
        5. 潜在挑战和缓解措施
        """
        
        return self.gemini.generate_response(prompt, temperature=0.8)

我们将 Semantic Kernel 逻辑封装在 SemanticKernelGeminiPlugin 中,初始化 Kernel 和 GeminiWrapper 来支持自定义 AI 函数。使用 @kernel_function 装饰器,我们声明了诸如 analyze_text、generate_summary、code_analysis 和 creative_solution 等方法,每个方法都构建结构化提示并将繁重的工作委托给 Gemini Flash。

构建高级多智能体系统

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class AdvancedGeminiAgent:
    """使用 Gemini Flash、AutoGen 和 Semantic Kernel 的高级 AI 智能体"""
    
    def __init__(self):
        self.sk_plugin = SemanticKernelGeminiPlugin()
        self.gemini = GeminiWrapper()
        self.setup_agents()
    
    def setup_agents(self):
        """使用 Gemini Flash 初始化 AutoGen 智能体"""
        
        gemini_config = {
            "config_list": [{"model": "gemini-1.5-flash", "api_key": GEMINI_API_KEY}],
            "temperature": 0.7,
        }
        
        # 主助手智能体
        self.assistant = autogen.ConversableAgent(
            name="GeminiAssistant",
            llm_config=gemini_config,
            system_message="""你是一个由 Gemini Flash 驱动的高级 AI 助手,具备 Semantic Kernel 能力。
            你擅长分析、问题解决和创意思维。始终提供全面、可操作的见解。
            使用结构化响应并考虑多个角度。""",
            human_input_mode="NEVER",
        )
        
        # 代码审查智能体
        self.code_reviewer = autogen.ConversableAgent(
            name="GeminiCodeReviewer",
            llm_config={**gemini_config, "temperature": 0.3},
            system_message="""你是一个由 Gemini Flash 驱动的高级代码审查员。
            分析代码的最佳实践、安全性、性能和可维护性。
            提供具体、可操作的反馈和示例。""",
            human_input_mode="NEVER",
        )
        
        # 创意分析师智能体
        self.creative_analyst = autogen.ConversableAgent(
            name="GeminiCreativeAnalyst",
            llm_config={**gemini_config, "temperature": 0.8},
            system_message="""你是一个由 Gemini Flash 驱动的创意问题解决者和创新专家。
            生成创新解决方案,提供新鲜视角。
            平衡创意与实用性。""",
            human_input_mode="NEVER",
        )
        
        # 数据专家智能体
        self.data_specialist = autogen.ConversableAgent(
            name="GeminiDataSpecialist",
            llm_config={**gemini_config, "temperature": 0.4},
            system_message="""你是一个由 Gemini Flash 驱动的数据分析专家。
            提供基于证据的建议和统计视角。""",
            human_input_mode="NEVER",
        )
        
        # 用户代理
        self.user_proxy = autogen.ConversableAgent(
            name="UserProxy",
            human_input_mode="NEVER",
            max_consecutive_auto_reply=2,
            is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
            llm_config=False,
        )
    
    def analyze_with_semantic_kernel(self, content: str, analysis_type: str) -> str:
        """AutoGen 和 Semantic Kernel 之间的桥接函数"""
        try:
            if analysis_type == "text":
                return self.sk_plugin.analyze_text(content)
            elif analysis_type == "code":
                return self.sk_plugin.code_analysis(content)
            elif analysis_type == "summary":
                return self.sk_plugin.generate_summary(content)
            elif analysis_type == "creative":
                return self.sk_plugin.creative_solution(content)
            else:
                return "无效的分析类型。请使用 'text'、'code'、'summary' 或 'creative'。"
        except Exception as e:
            return f"Semantic Kernel 分析错误: {str(e)}"
    
    def multi_agent_collaboration(self, task: str) -> Dict[str, str]:
        """使用 Gemini 编排多智能体协作"""
        results = {}
        
        agents = {
            "assistant": (self.assistant, "综合分析"),
            "code_reviewer": (self.code_reviewer, "代码审查视角"),
            "creative_analyst": (self.creative_analyst, "创意解决方案"),
            "data_specialist": (self.data_specialist, "数据驱动见解")
        }
        
        for agent_name, (agent, perspective) in agents.items():
            try:
                prompt = f"任务: {task}\n\n请从{perspective}的角度分析这个任务。"
                response = agent.generate_reply([{"role": "user", "content": prompt}])
                results[agent_name] = response if isinstance(response, str) else str(response)
            except Exception as e:
                results[agent_name] = f"智能体 {agent_name} 错误: {str(e)}"
        
        return results
    
    def run_comprehensive_analysis(self, query: str) -> Dict[str, Any]:
        """使用所有 Gemini 驱动的能力运行综合分析"""
        results = {}
        
        # Semantic Kernel 分析
        analyses = ["text", "summary", "creative"]
        for analysis_type in analyses:
            try:
                results[f"sk_{analysis_type}"] = self.analyze_with_semantic_kernel(query, analysis_type)
            except Exception as e:
                results[f"sk_{analysis_type}"] = f"错误: {str(e)}"
        
        # 多智能体协作
        try:
            results["multi_agent"] = self.multi_agent_collaboration(query)
        except Exception as e:
            results["multi_agent"] = f"多智能体错误: {str(e)}"
        
        # 直接 Gemini 调用
        try:
            results["direct_gemini"] = self.gemini.generate_response(
                f"请对以下内容提供综合分析: {query}", temperature=0.6
            )
        except Exception as e:
            results["direct_gemini"] = f"直接 Gemini 错误: {str(e)}"
        
        return results

主要演示函数

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def main():
    """Google Colab 中使用 Gemini Flash 的主执行函数"""
    print("🚀 初始化高级 Gemini Flash AI 智能体...")
    print("⚡ 使用 Gemini 1.5 Flash 进行高速、成本效益的 AI 处理")
    
    try:
        agent = AdvancedGeminiAgent()
        print("✅ 智能体初始化成功!")
    except Exception as e:
        print(f"❌ 初始化错误: {str(e)}")
        print("💡 请确保设置你的 Gemini API 密钥!")
        return
    
    demo_queries = [
        "AI 如何改变发展中国家的教育?",
        "def fibonacci(n): return n if n <= 1 else fibonacci(n-1) + fibonacci(n-2)",
        "2025年最有前景的可再生能源技术有哪些?"
    ]
    
    print("\n🔍 运行 Gemini Flash 驱动的分析...")
    
    for i, query in enumerate(demo_queries, 1):
        print(f"\n{'='*60}")
        print(f"🎯 演示 {i}: {query}")
        print('='*60)
        
        try:
            results = agent.run_comprehensive_analysis(query)
            
            for key, value in results.items():
                if key == "multi_agent" and isinstance(value, dict):
                    print(f"\n🤖 {key.upper().replace('_', ' ')}:")
                    for agent_name, response in value.items():
                        print(f"  👤 {agent_name}: {str(response)[:200]}...")
                else:
                    print(f"\n📊 {key.upper().replace('_', ' ')}:")
                    print(f"   {str(value)[:300]}...")
        
        except Exception as e:
            print(f"❌ 演示 {i} 错误: {str(e)}")
    
    print(f"\n{'='*60}")
    print("🎉 Gemini Flash AI 智能体演示完成!")
    print("💡 要使用你的 API 密钥,请替换 'your-gemini-api-key-here'")
    print("🔗 在此获取免费的 Gemini API 密钥: https://makersuite.google.com/app/apikey?WT.mc_id=AI-MVP-5003172")

if __name__ == "__main__":
    main()

总结

通过这个教程,我们展示了 AutoGen 和 Semantic Kernel 如何相互补充,创建一个由 Gemini Flash 驱动的多功能、多智能体 AI 系统。我们重点介绍了 AutoGen 如何简化不同专家智能体的编排,而 Semantic Kernel 为定义和调用高级 AI 函数提供了清晰的声明性层。

通过在 Colab 笔记本中整合这些工具,我们实现了复杂 AI 工作流的快速实验和原型设计,而不会牺牲清晰度或控制力。这种架构为构建能够处理多种任务类型的智能 AI 助手提供了强大的基础。