Google Gemini 3.5：原生多模态与智能体架构的范式革命

# Python示例：使用Gemini Omni进行原生多模态推理
import google.generativeai as genai

# 配置API
genai.configure(api_key="YOUR_API_KEY")

# 加载Gemini Omni模型
model = genai.GenerativeModel('gemini-omni-flash')

# 原生多模态输入：同时传入视频、音频和文本指令
video_path = "cycling_trip.mp4"
audio_path = "commentary.wav"

# 读取媒体文件
video_data = genai.upload_file(video_path)
audio_data = genai.upload_file(audio_path)

# 发送多模态指令
response = model.generate_content([
    video_data,
    audio_data,
    "请将视频中的户外骑行背景更换为雪地场景，并调整摄像机视角为侧面跟拍"
])

print(f"生成结果: {response.text}")
print(f"生成类型: {type(response.candidates[0].content)}")

// Go示例：使用Gemini SDK进行原生多模态调用
package main

import (
    "context"
    "fmt"
    "log"
    
    "github.com/google/generative-ai-go/genai"
    "google.golang.org/api/option"
)

func main() {
    ctx := context.Background()
    
    // 初始化Gemini客户端
    client, err := genai.NewClient(ctx, option.WithAPIKey("YOUR_API_KEY"))
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()
    
    // 加载Gemini Omni Flash模型
    model := client.GenerativeModel("gemini-omni-flash")
    
    // 创建多模态内容parts
    videoPart, err := genai.UploadFileFromPath(ctx, "cycling_trip.mp4")
    if err != nil {
        log.Fatal(err)
    }
    defer videoPart.Close(ctx)
    
    audioPart, err := genai.UploadFileFromPath(ctx, "commentary.wav")
    if err != nil {
        log.Fatal(err)
    }
    defer audioPart.Close(ctx)
    
    textPart := genai.Text("请将视频中的户外骑行背景更换为雪地场景")
    
    // 生成多模态内容
    resp, err := model.GenerateContent(ctx, videoPart, audioPart, textPart)
    if err != nil {
        log.Fatal(err)
    }
    
    fmt.Printf("生成结果: %s\n", resp.Candidates[0].Content.Parts[0].(genai.Text))
}

# Python示例：Gemini 3.5 Flash高速流式推理
import google.generativeai as genai
import time

genai.configure(api_key="YOUR_API_KEY")

model = genai.GenerativeModel('gemini-3.5-flash')

# 启用流式响应以获得更低延迟
prompt = """
请分析以下代码并提供性能优化建议：
"""

code_snippet = """
def process_large_dataset(data, batch_size=1000):
    results = []
    for i in range(0, len(data), batch_size):
        batch = data[i:i+batch_size]
        # 模拟数据处理
        processed = [transform(item) for item in batch]
        results.extend(processed)
    return results
"""

start_time = time.time()
chunk_count = 0

# 使用generate_content的stream参数
for chunk in model.generate_content(
    [prompt + code_snippet],
    generation_config=genai.types.GenerationConfig(
        max_output_tokens=2048,
        temperature=0.7
    ),
    stream=True  # 启用流式输出
):
    chunk_count += 1
    print(chunk.text, end='', flush=True)

elapsed = time.time() - start_time
print(f"\n\n📊 统计信息:")
print(f"   总耗时: {elapsed:.2f}秒")
print(f"   输出chunk数: {chunk_count}")
print(f"   平均每chunk耗时: {elapsed/chunk_count:.3f}秒")

# Python示例：构建代码智能体
from google.generativeai import GenerativeModel
from dataclasses import dataclass
from typing import List, Dict, Optional
import subprocess

@dataclass
class CodeAgentConfig:
    model_name: str = "gemini-3.5-flash"
    max_iterations: int = 10
    timeout_per_task: int = 60

class CodeAgent:
    """基于Gemini 3.5 Flash的代码智能体"""
    
    def __init__(self, api_key: str, config: CodeAgentConfig = None):
        import google.generativeai as genai
        genai.configure(api_key=api_key)
        self.config = config or CodeAgentConfig()
        self.model = GenerativeModel(self.config.model_name)
        self.conversation_history = []
    
    def execute_command(self, command: str) -> Dict[str, str]:
        """执行系统命令"""
        try:
            result = subprocess.run(
                command,
                shell=True,
                capture_output=True,
                text=True,
                timeout=self.config.timeout_per_task
            )
            return {
                "stdout": result.stdout,
                "stderr": result.stderr,
                "returncode": result.returncode
            }
        except subprocess.TimeoutExpired:
            return {"error": "Command timeout"}
    
    def solve_task(self, task_description: str) -> str:
        """解决编程任务"""
        prompt = f"""
        你是专业的代码智能体。请完成以下任务：
        
        任务：{task_description}
        
        请按照以下步骤执行：
        1. 分析任务需求
        2. 编写代码
        3. 执行并验证
        4. 修复任何错误
        
        最终输出：可运行的完整代码
        """
        
        for iteration in range(self.config.max_iterations):
            response = self.model.generate_content(prompt)
            code = response.text
            
            # 尝试执行代码
            exec_result = self.execute_command(code)
            
            if exec_result.get("returncode") == 0:
                return f"✅ 任务完成！\n\n执行结果:\n{exec_result.get('stdout', '')}"
            elif "error" in exec_result:
                prompt += f"\n\n上次执行失败: {exec_result['error']}\n请修复代码"
            else:
                prompt += f"\n\n执行结果:\n{exec_result.get('stderr', '')}\n请修复错误"
        
        return "❌ 任务失败：达到最大迭代次数"

# 使用示例
agent = CodeAgent(api_key="YOUR_API_KEY")
result = agent.solve_task(
    "创建一个Python脚本，读取当前目录下的所有CSV文件，"
    "合并它们，并计算每列的平均值后保存到merged_stats.csv"
)
print(result)

// Go示例：构建工作流智能体
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"

    "github.com/google/generative-ai-go/genai"
    "google.golang.org/api/option"
)

// TaskState represents the state of a workflow task
type TaskState struct {
    Status      string                 `json:"status"`
    Result      map[string]interface{} `json:"result,omitempty"`
    Error       string                 `json:"error,omitempty"`
    SubTasks    []TaskState            `json:"subtasks,omitempty"`
}

// WorkflowAgent handles complex multi-step workflows
type WorkflowAgent struct {
    client *genai.Client
    model  *genai.GenerativeModel
}

// NewWorkflowAgent creates a new workflow agent instance
func NewWorkflowAgent(ctx context.Context, apiKey string) (*WorkflowAgent, error) {
    client, err := genai.NewClient(ctx, option.WithAPIKey(apiKey))
    if err != nil {
        return nil, fmt.Errorf("failed to create client: %w", err)
    }
    
    return &WorkflowAgent{
        client: client,
        model:  client.GenerativeModel("gemini-3.5-flash"),
    }, nil
}

// DecomposeTask breaks down a complex task into subtasks
func (a *WorkflowAgent) DecomposeTask(ctx context.Context, task string) ([]string, error) {
    prompt := fmt.Sprintf(`
    请将以下复杂任务分解为可并行执行的子任务：
    
    任务：%s
    
    输出格式（JSON数组）：
    ["子任务1", "子任务2", "子任务3", ...]
    
    原则：
    - 每个子任务应该是独立的
    - 子任务之间尽量可以并行执行
    - 子任务描述要清晰明确
    `, task)

    resp, err := a.model.GenerateContent(ctx, genai.Text(prompt))
    if err != nil {
        return nil, err
    }

    var subtasks []string
    if err := json.Unmarshal([]byte(resp.Candidates[0].Content.Parts[0].(genai.Text)), &subtasks); err != nil {
        // Fallback to simple parsing
        return parseSubtasks(resp.Candidates[0].Content.Parts[0].(genai.Text))
    }
    
    return subtasks, nil
}

// ExecuteSubTask executes a single subtask
func (a *WorkflowAgent) ExecuteSubTask(ctx context.Context, subtask string) (*TaskState, error) {
    prompt := fmt.Sprintf(`
    请执行以下子任务，并提供详细的执行结果：
    
    子任务：%s
    
    请返回JSON格式的执行结果：
    {
        "status": "success|failed",
        "result": { ... },
        "error": "错误信息（如果有）"
    }
    `, subtask)

    resp, err := a.model.GenerateContent(ctx, genai.Text(prompt))
    if err != nil {
        return nil, err
    }

    var state TaskState
    if err := json.Unmarshal([]byte(resp.Candidates[0].Content.Parts[0].(genai.Text)), &state); err != nil {
        state = TaskState{
            Status: "success",
            Result: map[string]interface{}{"response": resp.Candidates[0].Content.Parts[0].(genai.Text)},
        }
    }

    return &state, nil
}

// ExecuteParallel runs multiple subtasks in parallel
func (a *WorkflowAgent) ExecuteParallel(ctx context.Context, subtasks []string) ([]TaskState, error) {
    results := make([]TaskState, len(subtasks))
    errors := make([]error, len(subtasks))

    // Use goroutines for parallel execution
    done := make(chan bool)
    
    for i, subtask := range subtasks {
        go func(idx int, task string) {
            state, err := a.ExecuteSubTask(ctx, task)
            if err != nil {
                errors[idx] = err
            } else {
                results[idx] = *state
            }
            done <- true
        }(i, subtask)
    }

    // Wait for all goroutines
    for i := 0; i < len(subtasks); i++ {
        <-done
    }

    // Check for errors
    for _, err := range errors {
        if err != nil {
            log.Printf("Subtask error: %v", err)
        }
    }

    return results, nil
}

// RunWorkflow executes the complete workflow
func (a *WorkflowAgent) RunWorkflow(ctx context.Context, task string) (*TaskState, error) {
    fmt.Printf("🔄 开始分解任务...\n")
    subtasks, err := a.DecomposeTask(ctx, task)
    if err != nil {
        return nil, fmt.Errorf("task decomposition failed: %w", err)
    }

    fmt.Printf("📋 分解为 %d 个子任务\n", len(subtasks))
    for i, st := range subtasks {
        fmt.Printf("   %d. %s\n", i+1, st)
    }

    fmt.Printf("⚡ 开始并行执行...\n")
    start := time.Now()
    results, err := a.ExecuteParallel(ctx, subtasks)
    elapsed := time.Since(start)

    if err != nil {
        return &TaskState{Status: "failed", Error: err.Error()}, nil
    }

    // Aggregate results
    finalResult := &TaskState{
        Status:   "success",
        SubTasks: results,
        Result: map[string]interface{}{
            "total_tasks":     len(subtasks),
            "execution_time":  elapsed.String(),
            "success_count":   countSuccess(results),
        },
    }

    fmt.Printf("✅ 工作流完成！耗时: %v\n", elapsed)
    return finalResult, nil
}

func countSuccess(states []TaskState) int {
    count := 0
    for _, s := range states {
        if s.Status == "success" {
            count++
        }
    }
    return count
}

func parseSubtasks(text string) ([]string, error) {
    // Simple fallback parser
    return []string{text}, nil
}

func main() {
    ctx := context.Background()
    
    agent, err := NewWorkflowAgent(ctx, "YOUR_API_KEY")
    if err != nil {
        log.Fatal(err)
    }
    defer agent.client.Close()
    
    result, err := agent.RunWorkflow(ctx, 
        "帮我完成数据分析和报告生成任务：1) 读取sales_data.csv 2) 计算月度汇总 3) 生成可视化图表 4) 导出PDF报告",
    )
    if err != nil {
        log.Fatal(err)
    }
    
    jsonResult, _ := json.MarshalIndent(result, "", "  ")
    fmt.Printf("\n📊 最终结果:\n%s\n", jsonResult)
}

# Python示例：使用CodeMender进行代码安全修复
import google.generativeai as genai
import re
from typing import List, Dict

genai.configure(api_key="YOUR_API_KEY")
model = genai.GenerativeModel('gemini-3.5-flash')

class CodeMender:
    """代码安全检查与修复工具"""
    
    # 常见安全漏洞模式
    VULNERABLE_PATTERNS = {
        'sql_injection': r'(SELECT|INSERT|UPDATE|DELETE).*\%s.*format|%\s*\)',
        'xss': r'(innerHTML|dangerouslySetInnerHTML)\s*=',
        'path_traversal': r'open\([^)]*[\+\.]\s*(request\.|os\.environ)',
        'hardcoded_secret': r'(password|secret|api_key|token)\s*=\s*["\'][^"\']{8,}["\']',
    }
    
    def __init__(self):
        self.vulnerabilities = []
    
    def scan_code(self, code: str) -> List[Dict]:
        """扫描代码中的安全漏洞"""
        results = []
        
        for vuln_type, pattern in self.VULNERABLE_PATTERNS.items():
            matches = re.finditer(pattern, code, re.IGNORECASE)
            for match in matches:
                results.append({
                    'type': vuln_type,
                    'line': code[:match.start()].count('\n') + 1,
                    'code': match.group(0),
                    'severity': self._get_severity(vuln_type)
                })
        
        return results
    
    def _get_severity(self, vuln_type: str) -> str:
        severities = {
            'sql_injection': 'CRITICAL',
            'xss': 'HIGH',
            'path_traversal': 'HIGH',
            'hardcoded_secret': 'MEDIUM'
        }
        return severities.get(vuln_type, 'LOW')
    
    def fix_vulnerability(self, code: str, vuln: Dict) -> str:
        """使用Gemini修复漏洞"""
        prompt = f"""
        请修复以下代码中的安全漏洞：
        
        漏洞类型：{vuln['type']}
        严重程度：{vuln['severity']}
        问题代码：
        {vuln['code']}
        
        完整代码上下文：
        {code}
        
        请提供修复后的代码，并解释修复方案。
        """
        
        response = model.generate_content(prompt)
        return response.text

# 使用示例
code_mender = CodeMender()

sample_vulnerable_code = '''
import sqlite3

def get_user_data(user_id):
    # SQL注入漏洞
    query = f"SELECT * FROM users WHERE id = {user_id}"
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()
    cursor.execute(query)
    return cursor.fetchall()

def render_comment(comment):
    # XSS漏洞
    return f"<div>{comment}</div>"

API_KEY = "sk-1234567890abcdef"  # 硬编码密钥
'''

vulnerabilities = code_mender.scan_code(sample_vulnerable_code)

print("🔍 安全扫描结果:")
print("=" * 50)
for i, vuln in enumerate(vulnerabilities, 1):
    print(f"\n[{i}] {vuln['severity']} - {vuln['type']}")
    print(f"    位置: 第{vuln['line']}行")
    print(f"    代码: {vuln['code']}")

print("\n" + "=" * 50)
print("🛠️ 开始自动修复...")

for vuln in vulnerabilities:
    fixed_code = code_mender.fix_vulnerability(sample_vulnerable_code, vuln)
    print(f"\n修复 {vuln['type']}:")
    print(fixed_code[:500] + "..." if len(fixed_code) > 500 else fixed_code)

# Python示例：使用Gemini Omni进行视频编辑控制
import google.generativeai as genai
from typing import List, Optional

genai.configure(api_key="YOUR_API_KEY")

class OmniVideoEditor:
    """基于Gemini Omni的视频编辑器"""
    
    def __init__(self, api_key: str):
        genai.configure(api_key=api_key)
        self.model = genai.GenerativeModel('gemini-omni-flash')
    
    def edit_video(
        self,
        video_path: str,
        instructions: List[str],
        preserve_aspects: Optional[List[str]] = None
    ) -> str:
        """
        执行视频编辑指令
        
        Args:
            video_path: 源视频路径
            instructions: 编辑指令列表
            preserve_aspects: 需要保留的元素（如角色动作）
        """
        # 上传视频
        video = genai.upload_file(video_path)
        
        # 构建编辑提示
        prompt = self._build_edit_prompt(instructions, preserve_aspects)
        
        # 执行编辑
        response = self.model.generate_content([
            video,
            prompt
        ])
        
        return response.text
    
    def _build_edit_prompt(self, instructions: List[str], preserve: List[str]) -> str:
        """构建编辑提示"""
        prompt_parts = [
            "你是一个专业的视频编辑AI。请根据以下指令编辑视频：\n"
        ]
        
        prompt_parts.append("【编辑指令】")
        for i, inst in enumerate(instructions, 1):
            prompt_parts.append(f"{i}. {inst}")
        
        if preserve:
            prompt_parts.append("\n【必须保留的元素】")
            for p in preserve:
                prompt_parts.append(f"- {p}")
        
        prompt_parts.append("""
        
        请输出：
        1. 详细的编辑计划
        2. 每一帧的修改说明
        3. 最终视频参数设置
        """)
        
        return "\n".join(prompt_parts)

# 使用示例
editor = OmniVideoEditor(api_key="YOUR_API_KEY")

instructions = [
    "将户外骑行背景更换为雪地场景",
    "将天气从晴天改为飘雪的冬季",
    "调整整体色调为冷色调",
    "在画面中添加雪地脚印"
]

preserve_aspects = [
    "骑行者的人物动作和服装保持不变",
    "自行车结构完整",
    "骑行速度保持一致"
]

result = editor.edit_video(
    video_path="cycling_trip.mp4",
    instructions=instructions,
    preserve_aspects=preserve_aspects
)

print("📹 视频编辑计划:")
print(result)

# Python示例：Antigravity SDK基础用法
from google.antigravity import Agent, AgentOrchestrator, ToolRegistry
import asyncio

# 定义自定义工具
class CustomTool:
    name = "data_processor"
    description = "处理和分析结构化数据"
    
    async def execute(self, params: dict) -> dict:
        data = params.get("data", [])
        operation = params.get("operation", "sum")
        
        if operation == "sum":
            return {"result": sum(data)}
        elif operation == "avg":
            return {"result": sum(data) / len(data)}
        elif operation == "max":
            return {"result": max(data)}
        
        return {"error": "Unknown operation"}

# 创建工具注册表
registry = ToolRegistry()
registry.register(CustomTool())

# 创建智能体
code_agent = Agent(
    name="code_writer",
    model="gemini-3.5-flash",
    role="Python代码编写专家",
    tools=["bash", "editor", "git"]
)

analysis_agent = Agent(
    name="data_analyst",
    model="gemini-3.5-flash", 
    role="数据分析专家",
    tools=["data_processor", "visualizer"]
)

# 创建编排器
orchestrator = AgentOrchestrator(
    agents=[code_agent, analysis_agent],
    max_parallel=5
)

# 执行复杂任务
async def main():
    result = await orchestrator.execute("""
        完成以下任务：
        1. 从API获取销售数据
        2. 进行数据分析并计算关键指标
        3. 生成可视化图表
        4. 将结果保存为报告
    """)
    
    print(f"任务状态: {result.status}")
    print(f"执行时间: {result.execution_time}")
    print(f"子任务数量: {len(result.subtask_results)}")

asyncio.run(main())

// Go示例：使用Antigravity Go SDK
package main

import (
    "context"
    "fmt"
    "log"
    
    "github.com/google/antigravity-go/agent"
    "github.com/google/antigravity-go/orchestrator"
)

func main() {
    ctx := context.Background()
    
    // 创建代码智能体
    codeAgent := agent.New(
        agent.WithName("code_writer"),
        agent.WithModel("gemini-3.5-flash"),
        agent.WithRole("Python代码编写专家"),
        agent.WithTools("bash", "editor", "git"),
    )
    
    // 创建数据分析智能体
    analysisAgent := agent.New(
        agent.WithName("data_analyst"),
        agent.WithModel("gemini-3.5-flash"),
        agent.WithRole("数据分析专家"),
        agent.WithTools("data_processor", "visualizer"),
    )
    
    // 创建编排器
    orch := orchestrator.New(
        orchestrator.WithAgents(codeAgent, analysisAgent),
        orchestrator.WithMaxParallel(5),
    )
    
    // 执行任务
    result, err := orch.Execute(ctx, `
        完成以下任务：
        1. 从API获取销售数据
        2. 进行数据分析并计算关键指标
        3. 生成可视化图表
        4. 将结果保存为报告
    `)
    if err != nil {
        log.Fatal(err)
    }
    
    fmt.Printf("任务状态: %s\n", result.Status)
    fmt.Printf("执行时间: %v\n", result.ExecutionTime)
    fmt.Printf("子任务数量: %d\n", len(result.SubtaskResults))
}

# Python示例：MCP协议集成
from mcp.client import MCPClient
from mcp.types import Tool, Resource, Prompt

# MCP客户端初始化
mcp_client = MCPClient(
    server_url="https://mcp.atlas.google.com/v1",
    api_key="YOUR_API_KEY"
)

@mcp_client.tool(name="web_search")
def web_search(query: str, max_results: int = 5) -> list:
    """
    执行网络搜索
    Args:
        query: 搜索关键词
        max_results: 最大结果数
    Returns:
        搜索结果列表
    """
    pass

@mcp_client.tool(name="file_operations")
def file_operations(
    operation: str,
    path: str,
    content: str = None
) -> dict:
    """
    文件操作工具
    Args:
        operation: 操作类型 (read/write/delete)
        path: 文件路径
        content: 写入内容（仅写操作）
    """
    pass

@mcp_client.tool(name="code_executor")
def code_executor(
    language: str,
    code: str,
    timeout: int = 60
) -> dict:
    """
    执行代码
    Args:
        language: 编程语言
        code: 代码内容
        timeout: 超时时间（秒）
    """
    pass

# 在智能体中使用MCP工具
async def research_agent(query: str):
    """研究型智能体"""
    
    # 使用MCP工具进行搜索
    search_results = await mcp_client.call_tool(
        "web_search",
        query=f"{query} 最新进展 2026",
        max_results=10
    )
    
    # 分析搜索结果
    analysis = model.generate_content(f"""
        分析以下搜索结果，提取关键信息：
        {search_results}
        
        输出结构化的分析报告。
    """)
    
    return analysis.text

# Python示例：开户合规检查智能体
import google.generativeai as genai
from typing import List, Dict, Optional
from dataclasses import dataclass
import re

genai.configure(api_key="YOUR_API_KEY")

@dataclass
class DocumentAnalysis:
    doc_type: str
    content_summary: str
    key_findings: List[str]
    risk_flags: List[str]
    compliance_score: float

class AccountOpeningAgent:
    """开户合规检查智能体"""
    
    def __init__(self):
        self.model = genai.GenerativeModel('gemini-3.5-flash')
        self.required_docs = [
            "身份证明",
            "地址证明", 
            "税务信息(W-9/W-8)",
            "资金来源证明",
            "风险评估问卷"
        ]
    
    async def analyze_document(self, doc_content: str, doc_type: str) -> DocumentAnalysis:
        """分析单个文档"""
        
        prompt = f"""
        请分析以下{doc_type}文档，提取关键信息：
        
        文档内容：
        {doc_content[:5000]}  # 限制长度
        
        请返回JSON格式的分析结果：
        {{
            "doc_type": "文档类型",
            "content_summary": "内容摘要",
            "key_findings": ["关键发现1", "关键发现2"],
            "risk_flags": ["风险标记1", "风险标记2"],
            "compliance_score": 0.0-1.0之间的合规分数
        }}
        """
        
        response = self.model.generate_content(prompt)
        
        # 解析JSON响应
        # ... 解析代码
        
        return DocumentAnalysis(
            doc_type=doc_type,
            content_summary="解析后的摘要",
            key_findings=["找到的发现"],
            risk_flags=["识别的风险"],
            compliance_score=0.95
        )
    
    async def process_application(
        self, 
        applicant_id: str,
        documents: Dict[str, str]
    ) -> Dict:
        """
        处理开户申请
        """
        results = {
            "applicant_id": applicant_id,
            "document_analysis": [],
            "missing_docs": [],
            "risk_assessment": None,
            "recommendation": None,
            "overall_compliance_score": 0.0
        }
        
        # 分析每个文档
        total_score = 0
        doc_count = 0
        
        for doc_type, content in documents.items():
            analysis = await self.analyze_document(content, doc_type)
            results["document_analysis"].append(analysis)
            total_score += analysis.compliance_score
            doc_count += 1
            
            # 检查风险标记
            if analysis.risk_flags:
                results["risk_assessment"] = results.get("risk_assessment", [])
                results["risk_assessment"].extend(analysis.risk_flags)
        
        # 检查缺失文档
        for required in self.required_docs:
            if required not in documents:
                results["missing_docs"].append(required)
        
        # 计算总体合规分数
        if doc_count > 0:
            results["overall_compliance_score"] = total_score / doc_count
        
        # 生成建议
        results["recommendation"] = self._generate_recommendation(results)
        
        return results
    
    def _generate_recommendation(self, results: Dict) -> str:
        """生成审批建议"""
        
        if results["missing_docs"]:
            return f"需要补充文档: {', '.join(results['missing_docs'])}"
        
        if results.get("risk_assessment"):
            if len(results["risk_assessment"]) > 3:
                return "拒绝 - 存在多项高风险标记"
            else:
                return "需要人工复核 - 存在风险标记"
        
        if results["overall_compliance_score"] >= 0.9:
            return "自动批准"
        elif results["overall_compliance_score"] >= 0.7:
            return "有条件批准 - 请在3天内补充说明"
        else:
            return "需要进一步审查"

# 使用示例
async def main():
    agent = AccountOpeningAgent()
    
    # 模拟开户申请文档
    application = {
        "身份证明": """
        姓名：张三
        身份证号：110101199001011234
        签发机关：北京市公安局
        有效期：2020-01-01 至 2030-01-01
        """,
        "地址证明": """
        住址：北京市朝阳区建国路88号
        物业费缴纳记录：2026年1-4月
        水电费账单：2026年3月
        """,
        "税务信息": """
        纳税人类型：个人
        W-9表格已提交
        纳税人识别号：91110105MA01234X5
        """,
        "资金来源证明": """
        职业：软件工程师
        年收入范围：50-80万
        首笔入金：50,000美元
        资金来源：工资收入
        """,
        "风险评估问卷": """
        投资经验：5年以上
        可承受风险等级：中高
        投资目标：资产增值
        对衍生品了解：有限
        """
    }
    
    result = await agent.process_application("APP-2026-001", application)
    
    print("📋 开户申请分析报告")
    print("=" * 60)
    print(f"申请人ID: {result['applicant_id']}")
    print(f"总体合规分数: {result['overall_compliance_score']:.2%}")
    print(f"\n建议: {result['recommendation']}")
    
    if result['missing_docs']:
        print(f"\n缺失文档: {', '.join(result['missing_docs'])}")
    
    if result.get('risk_assessment'):
        print(f"\n风险标记: {', '.join(result['risk_assessment'])}")

import asyncio
asyncio.run(main())

# Python示例：Shopify风格的商家增长分析系统
import asyncio
from typing import List, Dict
import google.generativeai as genai
from dataclasses import dataclass
from datetime import datetime, timedelta
import json

genai.configure(api_key="YOUR_API_KEY")

@dataclass
class MerchantData:
    merchant_id: str
    store_name: str
    sales_history: List[Dict]
    customer_data: List[Dict]
    inventory_data: List[Dict]
    marketing_campaigns: List[Dict]

class ParallelAnalysisAgent:
    """并行分析智能体系统"""
    
    def __init__(self):
        self.model = genai.GenerativeModel('gemini-3.5-flash')
        self.sub_agents = {
            'sales': SalesAnalysisAgent(self.model),
            'customer': CustomerAnalysisAgent(self.model),
            'inventory': InventoryAnalysisAgent(self.model),
            'marketing': MarketingAnalysisAgent(self.model)
        }
    
    async def analyze_merchant(self, data: MerchantData) -> Dict:
        """并行分析商家数据"""
        
        print(f"🚀 开始分析商家: {data.store_name}")
        
        # 并行启动所有子智能体
        tasks = [
            self.sub_agents['sales'].analyze(data.sales_history),
            self.sub_agents['customer'].analyze(data.customer_data),
            self.sub_agents['inventory'].analyze(data.inventory_data),
            self.sub_agents['marketing'].analyze(data.marketing_campaigns)
        ]
        
        # 并发执行
        results = await asyncio.gather(*tasks)
        
        sales_report, customer_report, inventory_report, marketing_report = results
        
        # 综合分析
        final_report = await self._synthesize_report(
            data,
            sales_report,
            customer_report,
            inventory_report,
            marketing_report
        )
        
        return final_report
    
    async def _synthesize_report(
        self,
        data: MerchantData,
        sales: Dict,
        customers: Dict,
        inventory: Dict,
        marketing: Dict
    ) -> Dict:
        """综合所有分析结果生成最终报告"""
        
        synthesis_prompt = f"""
        请综合以下四个维度的分析结果，为商家"{data.store_name}"生成增长建议：
        
        【销售分析】
        {json.dumps(sales, ensure_ascii=False, indent=2)}
        
        【客户分析】
        {json.dumps(customers, ensure_ascii=False, indent=2)}
        
        【库存分析】
        {json.dumps(inventory, ensure_ascii=False, indent=2)}
        
        【营销分析】
        {json.dumps(marketing, ensure_ascii=False, indent=2)}
        
        请生成包含以下内容的报告：
        1. 核心发现总结（不超过5点）
        2. 主要增长机会
        3. 具体行动计划（按优先级排序）
        4. 预期效果指标
        """
        
        response = self.model.generate_content(synthesis_prompt)
        
        return {
            "merchant_id": data.merchant_id,
            "store_name": data.store_name,
            "analysis_timestamp": datetime.now().isoformat(),
            "sales_insights": sales,
            "customer_insights": customers,
            "inventory_insights": inventory,
            "marketing_insights": marketing,
            "final_recommendations": response.text
        }


class SalesAnalysisAgent:
    """销售分析子智能体"""
    
    def __init__(self, model):
        self.model = model
    
    async def analyze(self, sales_data: List[Dict]) -> Dict:
        prompt = f"""
        分析以下销售数据，识别关键趋势和异常：
        
        数据范围: 最近90天
        数据条数: {len(sales_data)}
        
        请输出JSON格式的分析结果，包含：
        - total_revenue: 总营收
        - revenue_trend: 营收趋势 (up/down/stable)
        - top_products: 热销产品TOP5
        - peak_hours: 销售高峰时段
        - anomalies: 异常数据点
        """
        
        response = self.model.generate_content(prompt)
        return {"status": "completed", "analysis": "基于AI分析"}


class CustomerAnalysisAgent:
    """客户分析子智能体"""
    
    def __init__(self, model):
        self.model = model
    
    async def analyze(self, customer_data: List[Dict]) -> Dict:
        prompt = f"""
        分析以下客户数据，识别客户特征和价值：
        
        客户总数: {len(customer_data)}
        
        请输出JSON格式的分析结果，包含：
        - customer_segments: 客户分群
        - lifetime_value: 客户终身价值分布
        - churn_risk: 高流失风险客户比例
        - acquisition_cost: 平均获客成本
        - recommendations: 客户增长建议
        """
        
        response = self.model.generate_content(prompt)
        return {"status": "completed", "analysis": "基于AI分析"}


class InventoryAnalysisAgent:
    """库存分析子智能体"""
    
    def __init__(self, model):
        self.model = model
    
    async def analyze(self, inventory_data: List[Dict]) -> Dict:
        prompt = f"""
        分析以下库存数据，优化库存管理：
        
        SKU数量: {len(inventory_data)}
        
        请输出JSON格式的分析结果，包含：
        - stockout_rate: 缺货率
        - overstock_items: 积压商品
        - reorder_recommendations: 补货建议
        - turnover_rate: 库存周转率
        """
        
        response = self.model.generate_content(prompt)
        return {"status": "completed", "analysis": "基于AI分析"}


class MarketingAnalysisAgent:
    """营销分析子智能体"""
    
    def __init__(self, model):
        self.model = model
    
    async def analyze(self, campaign_data: List[Dict]) -> Dict:
        prompt = f"""
        分析以下营销数据，优化营销策略：
        
        活动数量: {len(campaign_data)}
        
        请输出JSON格式的分析结果，包含：
        - best_channels: 最佳营销渠道
        - roi_by_campaign: 各活动ROI
        - audience_insights: 受众洞察
        - budget_optimization: 预算优化建议
        """
        
        response = self.model.generate_content(prompt)
        return {"status": "completed", "analysis": "基于AI分析"}


# 使用示例
async def main():
    # 模拟商家数据
    sample_data = MerchantData(
        merchant_id="SHOP-001",
        store_name="时尚生活馆",
        sales_history=[{"date": "2026-05-01", "revenue": 5000} for _ in range(90)],
        customer_data=[{"id": f"C{i}", "ltv": 500 + i*50} for i in range(1000)],
        inventory_data=[{"sku": f"SKU-{i}", "stock": 100} for i in range(200)],
        marketing_campaigns=[{"name": "母亲节促销", "spend": 5000, "revenue": 25000}]
    )
    
    agent = ParallelAnalysisAgent()
    report = await agent.analyze_merchant(sample_data)
    
    print("\n📊 商家增长分析报告")
    print("=" * 60)
    print(f"商家: {report['store_name']}")
    print(f"分析时间: {report['analysis_timestamp']}")
    print(f"\n最终建议:\n{report['final_recommendations']}")

asyncio.run(main())