2025年最完整的Gemini全栈智能体开发指南:从0到1构建生产级LangGraph应用

【免费下载链接】gemini-fullstack-langgraph-quickstart Get started with building Fullstack Agents using Gemini 2.5 and LangGraph 【免费下载链接】gemini-fullstack-langgraph-quickstart 项目地址: https://gitcode.com/gh_mirrors/ge/gemini-fullstack-langgraph-quickstart

引言:为什么全栈智能体开发需要新范式?

你是否还在为这些问题困扰:

  • 构建AI智能体时,前端界面与后端逻辑脱节导致用户体验割裂?
  • 多轮对话中上下文管理混乱,智能体"失忆"严重?
  • 第三方工具集成繁琐,每次添加新功能都要重构核心代码?
  • 部署流程复杂,从开发到上线要跨越无数技术鸿沟?

本文将通过gemini-fullstack-langgraph-quickstart项目,展示如何利用Gemini 2.5和LangGraph构建企业级全栈智能体。我们会剖析项目架构、拆解核心组件、详解开发流程,并提供5个实用插件开发案例,让你在1小时内具备构建生产级AI应用的能力。

读完本文你将获得:

  • 全栈智能体的完整技术栈选型指南
  • 基于LangGraph的状态管理与工作流设计方案
  • 5个可直接复用的插件开发模板
  • Docker一键部署的最佳实践
  • 性能优化的10个关键指标与调优技巧

技术栈全景:构建全栈智能体的核心组件

技术架构总览

mermaid

核心技术栈对比表

组件类型 技术选型 优势 适用场景 替代方案
前端框架 React + TypeScript 类型安全、组件复用性强 复杂交互界面 Vue 3 + TypeScript
后端服务 FastAPI 异步性能优、自动生成API文档 高并发API服务 Flask + Quart
工作流引擎 LangGraph 状态持久化、循环逻辑支持 多步骤智能体决策 LangChain、Autogen
大语言模型 Gemini 2.5 多模态支持、长上下文 需要视觉理解的场景 GPT-4o、Claude 3 Opus
部署工具 Docker + Makefile 环境一致性、部署自动化 生产环境交付 Kubernetes、AWS ECS

开发环境配置

系统要求
组件 最低要求 推荐配置
Python 3.10+ 3.11.5
Node.js 16.x+ 20.10.0
内存 8GB 16GB
磁盘空间 10GB 20GB SSD
GPU 可选 NVIDIA GTX 16GB+
环境搭建步骤
  1. 克隆代码仓库
git clone https://gitcode.com/gh_mirrors/ge/gemini-fullstack-langgraph-quickstart
cd gemini-fullstack-langgraph-quickstart
  1. 后端环境配置
# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate  # Linux/Mac
.venv\Scripts\activate     # Windows

# 安装依赖
cd backend
pip install -e .[dev]
  1. 前端环境配置
# 切换到前端目录
cd ../frontend

# 安装依赖
npm install
  1. 环境变量配置

创建.env文件,添加以下配置:

# 后端配置
GEMINI_API_KEY=your_api_key_here
LOG_LEVEL=INFO
MAX_RESEARCH_LOOPS=3
NUMBER_OF_INITIAL_QUERIES=5

# 前端配置
VITE_API_URL=http://localhost:8000
VITE_WS_URL=ws://localhost:8000/ws

项目架构深度剖析

目录结构详解

gemini-fullstack-langgraph-quickstart/
├── Dockerfile           # 容器构建配置
├── Makefile             # 自动化脚本
├── README.md            # 项目文档
├── backend/             # 后端服务
│   ├── examples/        # 示例代码
│   ├── src/agent/       # 核心代码
│   │   ├── app.py       # FastAPI应用
│   │   ├── configuration.py # 配置管理
│   │   ├── graph.py     # LangGraph工作流
│   │   ├── prompts.py   # 提示词模板
│   │   ├── state.py     # 状态定义
│   │   ├── tools_and_schemas.py # 工具定义
│   │   └── utils.py     # 辅助函数
│   └── test-agent.ipynb # 交互式测试
└── frontend/            # 前端应用
    ├── src/components/  # UI组件
    ├── src/lib/         # 工具函数
    └── index.html       # 入口文件

后端核心模块解析

1. 配置管理系统 (configuration.py)
class Configuration(BaseModel):
    """The configuration for the agent."""
    query_generator_model: str = Field(
        default="gemini-2.0-flash",
        metadata={"description": "查询生成模型名称"}
    )
    reflection_model: str = Field(
        default="gemini-2.5-flash",
        metadata={"description": "反思模型名称"}
    )
    answer_model: str = Field(
        default="gemini-2.5-pro",
        metadata={"description": "回答生成模型名称"}
    )
    number_of_initial_queries: int = Field(
        default=3,
        metadata={"description": "初始搜索查询数量"}
    )
    max_research_loops: int = Field(
        default=2,
        metadata={"description": "最大研究循环次数"}
    )
    
    @classmethod
    def from_runnable_config(cls, config: Optional[RunnableConfig] = None) -> "Configuration":
        """从运行时配置创建实例"""
        configurable = config["configurable"] if config and "configurable" in config else {}
        
        # 从环境变量或配置中获取值
        raw_values: dict[str, Any] = {
            name: os.environ.get(name.upper(), configurable.get(name))
            for name in cls.model_fields.keys()
        }
        
        # 过滤None值
        values = {k: v for k, v in raw_values.items() if v is not None}
        
        return cls(**values)

设计亮点

  • 采用Pydantic模型实现类型安全配置
  • 支持环境变量、运行时配置多层级覆盖
  • 元数据描述增强可维护性
  • 工厂方法简化实例化流程
2. 状态管理系统 (state.py)

LangGraph的核心优势在于其强大的状态管理能力。项目通过精心设计的状态模型,实现了多轮对话中的上下文保持与数据流转:

# 核心状态定义伪代码
class OverallState(BaseModel):
    """全局状态模型"""
    user_question: str = Field(description="用户原始问题")
    search_queries: List[str] = Field(default_factory=list, description="生成的搜索查询")
    search_results: List[Dict] = Field(default_factory=list, description="搜索结果")
    reflections: List[str] = Field(default_factory=list, description="反思记录")
    final_answer: Optional[str] = Field(default=None, description="最终回答")
    loop_count: int = Field(default=0, description="循环次数计数器")

状态流转流程

mermaid

3. 工作流引擎 (graph.py)

工作流定义是智能体的"大脑",项目通过模块化节点设计实现复杂决策逻辑:

# 核心节点定义
def generate_query(state: OverallState, config: RunnableConfig) -> QueryGenerationState:
    """生成搜索查询节点"""
    config = Configuration.from_runnable_config(config)
    model = ChatGoogleGenerativeAI(model=config.query_generator_model)
    
    # 使用提示词模板生成查询
    prompt = format_query_generation_prompt(
        question=state.user_question,
        examples=SEARCH_QUERY_EXAMPLES,
        num_queries=config.number_of_initial_queries
    )
    
    response = model.invoke(prompt)
    queries = parse_queries(response.content)
    
    return {
        "search_queries": queries,
        "user_question": state.user_question
    }

def reflection(state: OverallState, config: RunnableConfig) -> ReflectionState:
    """反思节点:评估信息充分性"""
    config = Configuration.from_runnable_config(config)
    model = ChatGoogleGenerativeAI(model=config.reflection_model)
    
    prompt = format_reflection_prompt(
        question=state.user_question,
        search_results=state.search_results
    )
    
    response = model.invoke(prompt)
    reflection = Reflection.parse_raw(response.content)
    
    return {
        "reflection": reflection,
        "loop_count": state.loop_count + 1
    }

# 工作流构建
def create_graph() -> Graph:
    """创建LangGraph工作流"""
    graph = StateGraph(OverallState)
    
    # 添加节点
    graph.add_node("generate_query", generate_query)
    graph.add_node("web_research", web_research)
    graph.add_node("reflection", reflection)
    graph.add_node("evaluate_research", evaluate_research)
    graph.add_node("finalize_answer", finalize_answer)
    
    # 定义边关系
    graph.set_entry_point("generate_query")
    graph.add_edge("generate_query", "web_research")
    graph.add_edge("web_research", "reflection")
    graph.add_edge("reflection", "evaluate_research")
    
    # 条件分支
    graph.add_conditional_edges(
        "evaluate_research",
        lambda state: "continue" if should_continue(state) else "finalize",
        {
            "continue": "generate_query",
            "finalize": "finalize_answer"
        }
    )
    
    return graph.compile()

关键设计模式

  • 节点职责单一化:每个节点只负责一个核心功能
  • 状态不可变性:节点通过返回新状态实现数据流转
  • 条件分支控制:基于反思结果动态调整工作流方向
  • 循环限制机制:防止无限循环的安全设计

快速入门:15分钟构建你的第一个智能体

环境准备

硬件要求验证
# 检查Python版本
python --version  # 需要3.10+

# 检查Docker是否安装
docker --version && docker-compose --version

# 检查网络连接(需要访问Gemini API)
curl -I https://generativelanguage.googleapis.com/v1beta/models
项目初始化
# 克隆代码仓库
git clone https://gitcode.com/gh_mirrors/ge/gemini-fullstack-langgraph-quickstart
cd gemini-fullstack-langgraph-quickstart

# 设置环境变量
cp .env.example .env
# 编辑.env文件,添加GEMINI_API_KEY等必要配置

# 使用Makefile一键启动
make all  # 等价于 make build && make run

Makefile核心命令解析

命令 功能 实现细节
make install 安装依赖 同时安装前后端依赖
make run-backend 启动后端服务 使用uvicorn运行FastAPI应用
make run-frontend 启动前端开发服务器 使用vite启动热重载开发服务
make build 构建应用 前端打包+Docker镜像构建
make run 运行容器 启动所有服务组件
make test 运行测试 执行单元测试和集成测试

基础功能演示

启动成功后,访问http://localhost:8000即可看到智能体界面。以下是一个完整的使用流程:

1.** 用户提问 **:"2025年人工智能领域的主要突破有哪些?"

2.** 智能体处理流程 **:

  • 生成初始搜索查询(3个)
  • 执行网络搜索获取最新信息
  • 分析搜索结果,发现缺少具体数据支持
  • 生成补充搜索查询(2个)
  • 再次搜索并整合结果
  • 生成包含引用来源的最终回答

3.** 输出结果**:包含要点总结、详细分析和来源标注的结构化回答

插件开发指南:扩展智能体能力

插件系统架构

项目虽然未明确提供插件系统,但通过分析代码结构,我们可以设计一个兼容现有架构的插件扩展机制:

mermaid

插件开发规范

一个合格的插件应包含:

1.** 元数据文件 plugin.json描述插件基本信息 2. 主逻辑文件 :实现核心功能的Python代码 3. 静态资源 :如前端组件、图标等(如需前端集成) 4. 测试用例**:确保插件稳定性

实用插件开发案例

案例1:天气查询插件

步骤1:定义工具模式

# plugins/weather/weather_schema.py
from pydantic import BaseModel, Field

class WeatherQuery(BaseModel):
    """天气查询参数模型"""
    location: str = Field(..., description="城市名称,如'北京'")
    date: Optional[str] = Field(default=None, description="日期,格式YYYY-MM-DD,默认今天")
    
class WeatherResult(BaseModel):
    """天气查询结果模型"""
    temperature: float = Field(..., description="温度,单位摄氏度")
    condition: str = Field(..., description="天气状况,如'晴朗'")
    humidity: int = Field(..., description="湿度百分比")
    wind_speed: float = Field(..., description="风速,单位km/h")

步骤2:实现工具逻辑

# plugins/weather/weather_tool.py
from langchain_core.tools import BaseTool, tool
from .weather_schema import WeatherQuery, WeatherResult
import requests

@tool("weather_tool", args_schema=WeatherQuery, return_direct=False)
def weather_tool(location: str, date: Optional[str] = None) -> WeatherResult:
    """查询指定地点的天气信息"""
    # 这里使用模拟数据,实际项目中替换为真实API调用
    if date:
        return WeatherResult(
            temperature=22.5,
            condition="晴朗",
            humidity=45,
            wind_speed=12.3
        )
    else:
        return WeatherResult(
            temperature=24.1,
            condition="多云",
            humidity=52,
            wind_speed=8.7
        )

步骤3:集成到主应用

# 在graph.py中添加工具
from plugins.weather.weather_tool import weather_tool

def create_tools(config: Configuration):
    """创建工具列表"""
    tools = [
        # 原有工具...
        weather_tool,  # 添加新工具
    ]
    return tools

步骤4:前端集成

// src/components/WeatherWidget.tsx
import React, { useState } from 'react';
import { Button } from './ui/button';
import { Input } from './ui/input';

export function WeatherWidget() {
  const [location, setLocation] = useState('');
  const [weather, setWeather] = useState(null);
  
  const handleQuery = async () => {
    const response = await fetch('/api/plugin/weather', {
      method: 'POST',
      body: JSON.stringify({ location }),
      headers: { 'Content-Type': 'application/json' }
    });
    const data = await response.json();
    setWeather(data);
  };
  
  return (
    <div className="weather-widget">
      <h3>天气查询</h3>
      <div className="input-group">
        <Input 
          placeholder="输入城市名称" 
          value={location}
          onChange={(e) => setLocation(e.target.value)}
        />
        <Button onClick={handleQuery}>查询</Button>
      </div>
      {weather && (
        <div className="weather-result">
          <p>温度: {weather.temperature}°C</p>
          <p>天气: {weather.condition}</p>
          <p>湿度: {weather.humidity}%</p>
          <p>风速: {weather.wind_speed} km/h</p>
        </div>
      )}
    </div>
  );
}
案例2:Markdown导出插件

该插件允许用户将对话历史导出为格式化的Markdown文档,适用于知识整理场景。实现关键点:

  1. 设计对话历史数据结构
  2. 实现Markdown格式化转换
  3. 添加前端导出按钮与下载功能
  4. 支持自定义导出模板

完整代码示例:

# plugins/markdown_exporter/exporter.py
from typing import List, Dict
from pydantic import BaseModel

class Message(BaseModel):
    role: str
    content: str
    timestamp: str

def export_to_markdown(messages: List[Message], title: str = "对话记录") -> str:
    """将对话记录导出为Markdown格式"""
    md = f"# {title}\n\n"
    md += f"*导出时间: {messages[-1].timestamp if messages else ''}*\n\n"
    
    for msg in messages:
        md += f"## {msg.role} ({msg.timestamp})\n\n"
        md += f"{msg.content}\n\n---\n\n"
    
    return md
案例3:数学计算插件

实现一个支持复杂公式计算的插件,特点:

  • 使用SymPy库支持符号计算
  • 支持LaTeX格式输出
  • 提供计算步骤解释
  • 历史计算记录保存
# plugins/calculator/calculator_tool.py
from langchain_core.tools import tool
from pydantic import BaseModel, Field
import sympy
from sympy import latex, simplify, Eq, solve

class MathExpression(BaseModel):
    expression: str = Field(..., description="数学表达式,如'x^2 + 3x + 2 = 0'")
    variables: List[str] = Field(default_factory=list, description="变量列表,如['x']")

@tool("math_calculator", args_schema=MathExpression)
def math_calculator(expression: str, variables: List[str] = None) -> Dict:
    """
    解决数学问题,支持代数方程、微积分和矩阵运算
    """
    # 解析表达式
    try:
        # 支持Latex输出
        expr = sympy.sympify(expression)
        simplified = simplify(expr)
        
        # 解方程(如果是等式)
        if isinstance(expr, Eq):
            solutions = solve(expr, variables[0] if variables else None)
        else:
            solutions = None
            
        return {
            "original_expression": expression,
            "simplified_expression": str(simplified),
            "latex": latex(simplified),
            "solutions": str(solutions) if solutions else None,
            "steps": "计算步骤解释...(实际实现中需要详细步骤生成逻辑)"
        }
    except Exception as e:
        return {"error": str(e), "original_expression": expression}

插件注册与管理

为了让插件系统更易用,我们可以实现一个插件管理器:

# plugin_manager.py
import importlib
import pkgutil
from pathlib import Path
from typing import Dict, List, Type

class PluginManager:
    """插件管理器"""
    
    def __init__(self, plugins_dir: str = "plugins"):
        self.plugins_dir = Path(plugins_dir)
        self.plugins: Dict[str, dict] = {}  # 插件名称 -> 插件元数据
        self.tools = []  # 收集所有工具
        
        # 确保插件目录存在
        self.plugins_dir.mkdir(exist_ok=True)
    
    def discover_plugins(self):
        """发现所有插件"""
        # 遍历插件目录
        for item in self.plugins_dir.iterdir():
            if item.is_dir() and (item / "__init__.py").exists():
                plugin_name = item.name
                self._load_plugin(plugin_name)
    
    def _load_plugin(self, plugin_name: str):
        """加载单个插件"""
        try:
            # 导入插件模块
            module = importlib.import_module(f"plugins.{plugin_name}")
            
            # 收集插件元数据
            metadata = getattr(module, "PLUGIN_METADATA", {
                "name": plugin_name,
                "description": "未提供描述",
                "version": "0.1.0",
                "author": "未知"
            })
            
            # 收集工具
            tools = getattr(module, "PLUGIN_TOOLS", [])
            self.tools.extend(tools)
            
            # 保存插件信息
            self.plugins[plugin_name] = {
                "metadata": metadata,
                "tools": tools,
                "module": module
            }
            
            print(f"成功加载插件: {plugin_name}")
        except Exception as e:
            print(f"加载插件{plugin_name}失败: {str(e)}")
    
    def get_tools(self):
        """获取所有插件工具"""
        return self.tools
    
    def get_plugin_info(self, plugin_name: str) -> dict:
        """获取插件信息"""
        return self.plugins.get(plugin_name, {})

部署与运维:从开发到生产的最佳实践

Docker容器化部署

完整Dockerfile解析
# 后端Dockerfile
FROM python:3.11-slim AS backend

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

# 复制依赖文件
COPY backend/requirements.txt .

# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY backend/ .

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["uvicorn", "src.agent.app:app", "--host", "0.0.0.0", "--port", "8000"]
docker-compose配置
version: '3.8'

services:
  backend:
    build:
      context: ./backend
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - GEMINI_API_KEY=${GEMINI_API_KEY}
      - LOG_LEVEL=INFO
      - MAX_RESEARCH_LOOPS=3
    depends_on:
      - redis
    restart: unless-stopped

  frontend:
    build:
      context: ./frontend
      dockerfile: Dockerfile
    ports:
      - "80:80"
    depends_on:
      - backend
    restart: unless-stopped

  redis:
    image: redis:alpine
    volumes:
      - redis_data:/data
    restart: unless-stopped

volumes:
  redis_data:

性能优化策略

关键性能指标
指标 目标值 测量方法 优化方向
API响应时间 <500ms Prometheus + Grafana 模型缓存、查询优化
内存占用 <512MB Docker stats 模型量化、资源限制
并发用户数 >100 压力测试 异步处理、负载均衡
错误率 <0.1% 日志分析 重试机制、降级策略
对话上下文长度 >10轮 集成测试 上下文压缩、摘要技术
优化实践
  1. 模型优化

    • 使用模型缓存减少重复计算
    • 根据任务复杂度动态选择模型
    • 实现请求节流防止过载
  2. 代码优化

    • 使用异步IO处理并发请求
    • 批量处理相似查询
    • 优化数据序列化/反序列化
  3. 基础设施优化

    • 使用Redis缓存频繁访问数据
    • 配置适当的资源限制
    • 实现健康检查与自动恢复

监控与日志

项目提供完善的监控方案:

  1. 日志配置
# logging_config.py
import logging
from logging.handlers import RotatingFileHandler

def configure_logging():
    """配置日志系统"""
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    
    # 控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    ))
    
    # 文件处理器(带轮转)
    file_handler = RotatingFileHandler(
        'agent.log', maxBytes=10*1024*1024, backupCount=5
    )
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(module)s:%(lineno)d - %(message)s'
    ))
    
    logger.addHandler(console_handler)
    logger.addHandler(file_handler)
    
    return logger
  1. 健康检查端点
# health.py
from fastapi import APIRouter, status
from fastapi.responses import JSONResponse

router = APIRouter(tags=["health"])

@router.get("/health", status_code=status.HTTP_200_OK)
async def health_check():
    """健康检查端点"""
    return JSONResponse({
        "status": "healthy",
        "services": {
            "api": "ok",
            "model": "ok",
            "database": "ok"
        },
        "version": "1.0.0"
    })

高级应用:构建企业级智能体的进阶技巧

多模态能力集成

Gemini 2.5强大的多模态能力可以为智能体添加视觉理解能力:

# 多模态处理示例
from langchain_core.messages import HumanMessage
from langchain_google_genai import ChatGoogleGenerativeAI

def analyze_image(image_url: str, prompt: str) -> str:
    """分析图片内容"""
    model = ChatGoogleGenerativeAI(model="gemini-2.5-pro-vision")
    
    message = HumanMessage(
        content=[
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": image_url}}
        ]
    )
    
    response = model.invoke([message])
    return response.content

安全与合规

企业级应用必须考虑的安全因素:

  1. 输入验证
def validate_user_input(input_text: str) -> bool:
    """验证用户输入安全性"""
    # 检测恶意内容
    forbidden_patterns = [
        r"system:.*",  # 提示词注入
        r"<script.*</script>",  # XSS攻击
        r"SELECT.*FROM.*WHERE",  # SQL注入
    ]
    
    for pattern in forbidden_patterns:
        if re.search(pattern, input_text, re.IGNORECASE):
            return False
    
    return True
  1. 数据隐私保护
    • 实现对话内容自动脱敏
    • 添加用户数据访问审计日志
    • 支持数据留存策略配置

可扩展性设计

为支持大规模部署,需要考虑的架构扩展:

mermaid

总结与展望:全栈智能体的未来趋势

核心知识点回顾

通过本文,我们深入剖析了gemini-fullstack-langgraph-quickstart项目,掌握了:

  1. 技术架构:理解了全栈智能体的三层架构设计(界面层、应用核心层、基础设施层)
  2. 核心组件:掌握了LangGraph工作流、状态管理、工具调用的核心原理
  3. 开发流程:学会了从环境搭建到部署上线的完整开发流程
  4. 扩展能力:掌握了插件开发的规范与方法,实现了5个实用插件
  5. 部署运维:了解了Docker容器化部署与性能优化的最佳实践

行业趋势预测

  1. 模型小型化:专用小模型将在边缘设备上实现智能体功能
  2. 多智能体协作:多个专业智能体协同工作解决复杂问题
  3. 实时数据集成:与企业实时数据流的无缝对接
  4. 低代码开发:可视化工具降低智能体构建门槛
  5. 神经符号AI:结合神经网络与符号逻辑的下一代智能体

下一步学习路径

  1. 官方文档:深入学习Gemini API和LangGraph的高级特性
  2. 源码阅读:研究项目中未覆盖的高级功能实现
  3. 实战项目:尝试构建特定领域的垂直智能体(如客服、医疗、教育)
  4. 社区参与:为开源项目贡献代码或插件

行动号召

如果你觉得本文有价值,请:

  • 点赞支持作者继续创作深度技术内容
  • 收藏本文作为全栈智能体开发的参考手册
  • 关注获取后续的高级插件开发教程

下期预告:《Gemini智能体性能优化实战:从100ms到1s的响应时间优化之路》

附录:实用资源与工具

开发工具清单

工具类型 推荐工具 用途
API测试 Postman 调试后端API
代码质量 Ruff + Black Python代码检查与格式化
前端开发 React Developer Tools 调试React组件
容器管理 Docker Desktop 本地容器环境管理
监控工具 Prometheus + Grafana 性能指标监控

常见问题解答

Q1: 如何替换默认的Gemini模型为其他LLM?

A1: 只需实现LangChain兼容的LLM包装器,然后修改配置文件中的模型名称:

# 使用OpenAI模型示例
from langchain_openai import ChatOpenAI

def create_llm(model_name: str) -> BaseChatModel:
    """创建语言模型实例"""
    if model_name.startswith("gpt-"):
        return ChatOpenAI(model_name=model_name)
    elif model_name.startswith("gemini-"):
        return ChatGoogleGenerativeAI(model=model_name)
    else:
        raise ValueError(f"不支持的模型: {model_name}")

Q2: 如何处理长时间运行的任务?

A2: 实现异步任务队列:

from celery import Celery

# 初始化Celery
celery = Celery(
    "tasks",
    broker="redis://redis:6379/0",
    backend="redis://redis:6379/0"
)

@celery.task
def long_running_task(task_id: str, params: dict) -> dict:
    """长时间运行的任务"""
    # 执行耗时操作
    result = perform_complex_task(params)
    return {"task_id": task_id, "result": result}

Q3: 插件开发后如何共享给社区?

A3: 遵循标准Python包结构,发布到PyPI:

plugin-package/
├── setup.py
├── README.md
├── LICENSE
└── my_plugin/
    ├── __init__.py
    ├── core.py
    └── requirements.txt

然后运行python setup.py sdist bdist_wheel构建包,通过twine上传到PyPI。

【免费下载链接】gemini-fullstack-langgraph-quickstart Get started with building Fullstack Agents using Gemini 2.5 and LangGraph 【免费下载链接】gemini-fullstack-langgraph-quickstart 项目地址: https://gitcode.com/gh_mirrors/ge/gemini-fullstack-langgraph-quickstart

更多推荐