AgentScope模型集成：OpenAI、DashScope、Gemini支持

AgentScope作为新一代多智能体开发框架，其核心优势之一就是强大的模型无关性（Model Agnostic）设计。开发者只需编写一次代码，即可无缝切换使用OpenAI、DashScope、Gemini等主流大语言模型。本文将深入解析AgentScope的模型集成机制，展示如何高效利用不同模型提供商的能力。## 模型集成架构设计AgentScope采用统一的抽象接口设计，所有模型都继承...

石淞畅Oprah

1569人浏览 · 2025-09-01 16:29:37

石淞畅Oprah · 2025-09-01 16:29:37 发布

AgentScope模型集成：OpenAI、DashScope、Gemini支持

【免费下载链接】agentscope 项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope

概述

模型集成架构设计

AgentScope采用统一的抽象接口设计，所有模型都继承自ChatModelBase基类，确保接口一致性：

mermaid

支持的模型提供商对比

特性	OpenAI	DashScope	Gemini	Anthropic	Ollama
流式支持	✅	✅	✅	✅	✅
工具调用	✅	✅	✅	✅	✅
多模态	✅	✅	✅	✅	❌
推理模型	✅	✅	✅	✅	❌
异步调用	✅	✅	✅	✅	✅

统一接口设计

所有模型类都遵循相同的调用接口：

async def __call__(
    self,
    messages: list[dict],
    tools: list[dict] | None = None,
    tool_choice: Literal["auto", "none", "any", "required"] | str | None = None,
    structured_model: Type[BaseModel] | None = None,
    **config_kwargs: Any,
) -> ChatResponse | AsyncGenerator[ChatResponse, None]

模型初始化示例

OpenAI模型配置

from agentscope.model import OpenAIChatModel
import os

openai_model = OpenAIChatModel(
    model_name="gpt-4-turbo",
    api_key=os.environ["OPENAI_API_KEY"],
    stream=True,
    reasoning_effort="medium",
    organization="your-org-id"
)

DashScope模型配置

from agentscope.model import DashScopeChatModel

dashscope_model = DashScopeChatModel(
    model_name="qwen-max",
    api_key=os.environ["DASHSCOPE_API_KEY"],
    stream=True,
    enable_thinking=True,
    base_http_api_url=None
)

Gemini模型配置

from agentscope.model import GeminiChatModel

gemini_model = GeminiChatModel(
    model_name="gemini-1.5-pro",
    api_key=os.environ["GEMINI_API_KEY"],
    stream=True,
    thinking_config={"enabled": True},
    client_args={"timeout": 30}
)

统一响应格式

所有模型返回统一的ChatResponse对象：

response = ChatResponse(
    content=[
        ThinkingBlock(type="thinking", thinking="推理过程"),
        TextBlock(type="text", text="生成的文本"),
        ToolUseBlock(type="tool_use", id="tool_id", name="tool_name", input={}),
    ],
    identity="model_name",
    created=datetime.now(),
    usage={"prompt_tokens": 100, "completion_tokens": 50}
)

流式处理实现

AgentScope的流式处理采用累积式设计，每个chunk包含之前所有内容：

async def stream_example():
    model = OpenAIChatModel(
        model_name="gpt-4-turbo",
        api_key=os.environ["OPENAI_API_KEY"],
        stream=True
    )
    
    generator = await model(
        messages=[{"role": "user", "content": "请逐步解释..."}]
    )
    
    async for chunk in generator:
        # 每个chunk包含累积内容
        print(f"当前内容: {chunk.get_text_content()}")

工具调用统一处理

不同模型的工具调用格式差异被统一处理：

# 统一的工具JSON schema格式
tools_schema = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "网页搜索",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                },
                "required": ["query"]
            }
        }
    }
]

# 统一的工具选择模式
tool_choice_options = ["auto", "none", "any", "required"]

多模型切换策略

环境变量配置模式

import os
from agentscope.model import OpenAIChatModel, DashScopeChatModel

def get_model():
    provider = os.getenv("MODEL_PROVIDER", "openai")
    
    if provider == "openai":
        return OpenAIChatModel(
            model_name=os.getenv("OPENAI_MODEL", "gpt-4-turbo"),
            api_key=os.environ["OPENAI_API_KEY"]
        )
    elif provider == "dashscope":
        return DashScopeChatModel(
            model_name=os.getenv("DASHSCOPE_MODEL", "qwen-max"),
            api_key=os.environ["DASHSCOPE_API_KEY"]
        )

工厂模式实现

from typing import Union
from agentscope.model import OpenAIChatModel, DashScopeChatModel, GeminiChatModel

class ModelFactory:
    @staticmethod
    def create_model(provider: str, **kwargs) -> Union[OpenAIChatModel, DashScopeChatModel, GeminiChatModel]:
        config = {
            "model_name": kwargs.get("model_name"),
            "api_key": kwargs.get("api_key"),
            "stream": kwargs.get("stream", True)
        }
        
        if provider == "openai":
            return OpenAIChatModel(**config)
        elif provider == "dashscope":
            return DashScopeChatModel(**config)
        elif provider == "gemini":
            return GeminiChatModel(**config)
        else:
            raise ValueError(f"不支持的模型提供商: {provider}")

性能优化建议

连接池管理

import httpx
from agentscope.model import OpenAIChatModel

# 使用连接池提升性能
client_args = {
    "http_client": httpx.AsyncClient(
        limits=httpx.Limits(max_keepalive_connections=10, max_connections=20),
        timeout=30.0
    )
}

model = OpenAIChatModel(
    model_name="gpt-4-turbo",
    api_key=os.environ["OPENAI_API_KEY"],
    client_args=client_args
)

批量处理优化

async def batch_process(messages_list):
    model = DashScopeChatModel(
        model_name="qwen-max",
        api_key=os.environ["DASHSCOPE_API_KEY"]
    )
    
    # 并行处理多个请求
    tasks = [model(messages) for messages in messages_list]
    results = await asyncio.gather(*tasks)
    return results

错误处理与重试机制

from tenacity import retry, stop_after_attempt, wait_exponential

class RobustModel:
    def __init__(self, model):
        self.model = model
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10)
    )
    async def call_with_retry(self, messages, **kwargs):
        try:
            return await self.model(messages, **kwargs)
        except Exception as e:
            if "rate limit" in str(e).lower():
                raise
            # 其他错误进行重试
            raise

实际应用场景

多模型A/B测试

async def model_ab_test(prompt):
    models = [
        OpenAIChatModel(model_name="gpt-4-turbo", api_key=os.environ["OPENAI_API_KEY"]),
        DashScopeChatModel(model_name="qwen-max", api_key=os.environ["DASHSCOPE_API_KEY"]),
        GeminiChatModel(model_name="gemini-1.5-pro", api_key=os.environ["GEMINI_API_KEY"])
    ]
    
    results = {}
    for i, model in enumerate(models):
        response = await model([{"role": "user", "content": prompt}])
        results[f"model_{i}"] = response.get_text_content()
    
    return results

故障转移策略

class FallbackModel:
    def __init__(self, primary_model, fallback_models):
        self.primary = primary_model
        self.fallbacks = fallback_models
    
    async def call(self, messages, **kwargs):
        try:
            return await self.primary(messages, **kwargs)
        except Exception:
            for fallback in self.fallbacks:
                try:
                    return await fallback(messages, **kwargs)
                except Exception:
                    continue
            raise Exception("所有模型调用失败")

最佳实践总结

环境隔离：使用环境变量管理不同环境的API密钥和配置
统一接口：通过工厂模式封装模型创建逻辑
性能监控：实现调用统计和性能指标收集
错误恢复：设计完善的重试和故障转移机制
成本控制：根据使用场景选择合适的模型规格

扩展性设计

AgentScope的模型架构支持轻松扩展新的模型提供商：

from agentscope.model import ChatModelBase

class CustomChatModel(ChatModelBase):
    def __init__(self, model_name: str, stream: bool = True, **kwargs):
        super().__init__(model_name, stream)
        # 自定义初始化逻辑
    
    async def __call__(self, messages, tools=None, tool_choice=None, **kwargs):
        # 实现自定义调用逻辑
        pass
    
    def _format_tools_json_schemas(self, schemas):
        # 自定义工具格式转换
        pass

通过这种设计，开发者可以快速集成新的模型服务，同时保持与现有代码的兼容性。

AgentScope的模型集成架构为多智能体应用提供了强大的灵活性和可扩展性，让开发者能够专注于业务逻辑而非底层模型差异。

【免费下载链接】agentscope 项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope