CAMEL 多智能体框架接入 Azure OpenAI:AzureOpenAIModel 统一模型后端与 Responses API 实战指南
2026/9/14 17:09:51 网站建设 项目流程

CAMEL 多智能体框架接入 Azure OpenAI:AzureOpenAIModel 统一模型后端与 Responses API 实战指南

【免费下载链接】camel🐫 CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel

本篇技术指南以 CAMEL 仓库中的 AzureOpenAIModel 参考文档 为主体,结合 源码实现、官方示例与单元测试,系统讲解如何在 CAMEL 中通过统一模型后端接入 Azure OpenAI 服务。读完本文,你将掌握 Azure 环境变量配置、AzureOpenAIModel全部构造参数的含义与取舍、Chat Completions 与 Responses 两种 API 模式的切换,以及如何把 Azure 模型接入ChatAgent、工具调用与结构化输出等完整实战链路。

AzureOpenAIModel:把 Azure OpenAI 纳入 CAMEL 统一模型接口

AzureOpenAIModel是 CAMEL 多智能体框架为 Azure OpenAI 服务提供的模型后端实现,其类定义与定位在参考文档中表述为:

class AzureOpenAIModel(BaseModelBackend):

它把 Azure OpenAI API 封装进 CAMEL 统一的BaseModelBackend接口,使上层ChatAgentWorkforceTask等组件无需关心底层是 OpenAI 直连还是 Azure 部署。

从源码看(camel/models/azure_openai_model.py),该类直接继承自OpenAIModel

Inherits all Responses API, chat completion, and streaming methods fromOpenAIModel. Only the client initialization and Azure-specific configuration are different.

这意味着OpenAIModel提供的 Responses API、Chat Completion、流式推理、结构化输出等全部能力都被继承复用,AzureOpenAIModel只需覆盖两块差异化逻辑:

  1. 客户端初始化:使用openai.AzureOpenAI/openai.AsyncAzureOpenAI,并注入 Azure 特有的azure_endpointazure_deploymentapi_versionazure_ad_token等参数;
  2. Azure 特有配置:如api_mode"chat_completions""responses")、弃用参数兼容逻辑等。

因此,在 CAMEL 中通过ModelFactory创建模型时,只需要指定平台类型为ModelPlatformType.AZURE,工厂内部就会自动映射到本类。camel/models/model_factory.py 中注册了ModelPlatformType.AZURE: AzureOpenAIModel的对应关系,这是官方推荐的创建入口(下文详述)。

部署前提与环境变量

使用 Azure OpenAI 前,你需要在 Azure 门户中完成模型部署并获取三类信息:资源终结点(endpoint)API KeyAPI 版本号。CAMEL 通过环境变量与构造参数双重途径读取这些配置,对应关系如下表:

环境变量对应构造参数说明
AZURE_OPENAI_API_KEYapi_keyAzure OpenAI 资源的 API Key,必填
AZURE_OPENAI_BASE_URLurl资源终结点,形如https://<resource>.openai.azure.com/
AZURE_API_VERSIONapi_versionAPI 版本号,必填(详见下文校验逻辑),如2024-12-01-preview
AZURE_AD_TOKENazure_ad_tokenAzure Active Directory 令牌,使用 Entra ID 认证时设置
MODEL_TIMEOUTtimeout请求超时秒数,默认180
AZURE_DEPLOYMENT_NAMEmodel_type已弃用,见下文说明

这些环境变量在仓库的 .env.example 中有明确注释模板,可按from dotenv import load_dotenv; load_dotenv()的方式加载:

# Azure OpenAI API (https://azure.microsoft.com/products/cognitive-services/openai-service/) export AZURE_OPENAI_API_KEY="Fill your API key here" export AZURE_API_VERSION="Fill your API Version here" export AZURE_DEPLOYMENT_NAME="Fill your Deployment Name here" export AZURE_OPENAI_BASE_URL="Fill your Base URL here"

值得注意的是:参考文档与源码都指出AZURE_DEPLOYMENT_NAME环境变量已进入弃用流程。源码在__init__中会检查该变量并发出DeprecationWarning,提示改用model_type参数(camel/models/azure_openai_model.py)。因此新代码请把部署名直接作为model_type传入

api_version 的强制校验

在 源码初始化逻辑 中,api_version的解析顺序是:构造参数api_version→ 环境变量AZURE_API_VERSION。如果两者都未提供,会直接抛出异常:

if self.api_version is None: raise ValueError( "Must provide either the `api_version` argument " "or `AZURE_API_VERSION` environment variable." )

这是 Azure 与 OpenAI 直连最大的差异点:Azure 要求每个请求都携带 API 版本号,示例代码中普遍使用"2024-12-01-preview""2025-03-01-preview",请以你实际部署时选择的版本为准。

构造函数参数详解

参考文档完整列出了AzureOpenAIModel.__init__的签名与全部参数,现结合源码逐一展开:

def __init__( self, model_type: Union[ModelType, str], model_config_dict: Optional[Dict[str, Any]] = None, api_key: Optional[str] = None, url: Optional[str] = None, timeout: Optional[float] = None, token_counter: Optional[BaseTokenCounter] = None, api_version: Optional[str] = None, azure_ad_token_provider: Optional['AzureADTokenProvider'] = None, azure_ad_token: Optional[str] = None, max_retries: int = 3, client: Optional[Any] = None, async_client: Optional[Any] = None, azure_deployment_name: Optional[str] = None, **kwargs: Any ):
参数类型默认值含义与要点
model_typeUnion[ModelType, str]必填要创建后端的模型,必须是你在 Azure 中部署模型时选择的部署名(deployment name),而非模型 ID。可传ModelType枚举(如ModelType.GPT_4O)或字符串(如"gpt-4.1"
model_config_dictOptional[Dict[str, Any]]None会透传给openai.ChatCompletion.create()的配置字典;为None时使用ChatGPTConfig().as_dict()
api_keyOptional[str]None认证用的 API Key,缺省回落到AZURE_OPENAI_API_KEY环境变量
urlOptional[str]None服务地址,缺省回落到AZURE_OPENAI_BASE_URL环境变量
timeoutOptional[float]NoneAPI 调用超时(秒)。缺省时读取MODEL_TIMEOUT环境变量,最终默认180
token_counterOptional[BaseTokenCounter]None自定义 token 计数器;不提供时使用OpenAITokenCounter
api_versionOptional[str]NoneAzure API 版本,缺省读AZURE_API_VERSION,仍为空则抛ValueError
azure_ad_tokenOptional[str]NoneAzure Active Directory 令牌(Entra ID 认证场景),缺省读AZURE_AD_TOKEN
azure_ad_token_providerOptional[AzureADTokenProvider]None一个返回 AD 令牌的函数,每次请求都会调用;适合令牌自动刷新的场景
max_retriesint3API 调用的最大重试次数
clientOptional[Any]None自定义同步AzureOpenAI客户端实例,提供后不再自建。要求实现.chat.completions.create().beta.chat.completions.parse()接口;典型场景是 AReaL、rLLM 等 RL 框架提供的 Azure OpenAI 兼容客户端
async_clientOptional[Any]None自定义异步AsyncAzureOpenAI客户端实例,语义同上
azure_deployment_nameOptional[str]None已弃用,仅为向后兼容保留,未来版本将移除;使用时会发出DeprecationWarning且参数被忽略
**kwargsAny透传给客户端初始化的额外参数;提供自定义客户端时被忽略

参考文档还额外说明了继承而来的api_mode参数(源码签名为api_mode: Literal["chat_completions", "responses"] = "chat_completions"):

  • "chat_completions"(默认):走传统的 Chat Completions 接口;
  • "responses":走 Azure OpenAI Responses API,支持通过previous_response_id进行有状态响应链式调用。

非法值会直接抛错(ValueError: api_mode must be 'chat_completions' or 'responses'),该行为有单元测试覆盖(见下文)。

底层初始化行为(源码级)

__init__的执行链路大致如下(camel/models/azure_openai_model.py):

  1. 检查并警告弃用的azure_deployment_name参数与AZURE_DEPLOYMENT_NAME环境变量;
  2. 校验api_mode合法性,并初始化 Responses API 链式状态字典(_responses_previous_response_id_by_session_responses_last_message_count_by_session);
  3. model_config_dictNone时回退到ChatGPTConfig().as_dict()
  4. 依次从参数/环境变量解析api_keyurltimeout
  5. 跳过OpenAIModel.__init__(它要求OPENAI_API_KEY),直接调用BaseModelBackend.__init__完成公共初始化;
  6. 解析api_version并强制校验;
  7. 若已安装 langfuse 且未传自定义客户端,会优先使用langfuse.openai.AzureOpenAI/AsyncAzureOpenAI构造可观测客户端(实现自动链路追踪);否则构造原生AzureOpenAI/AsyncAzureOpenAI客户端,azure_endpointurlazure_deploymentmodel_type,并透传api_versionapi_key、AD 令牌、timeoutmax_retries

此外,参考文档列出的_sanitize_config_adapt_messages_for_o1_models在 Azure 实现中均为 no-op(直接原样返回),源码注释明确说明"Azure does not need"——Azure 侧无需对配置做清理或对 o1 系消息做适配。

快速上手:通过 ModelFactory 创建模型并驱动 ChatAgent

官方推荐通过ModelFactory.create创建模型,参考仓库中的 examples/models/azure_openai_model_example.py:

import os from camel.agents import ChatAgent from camel.models import ModelFactory from camel.types import ModelPlatformType # 请先设置以下环境变量: # export AZURE_OPENAI_BASE_URL="" # export AZURE_API_VERSION="" # export AZURE_OPENAI_API_KEY="" model = ModelFactory.create( model_platform=ModelPlatformType.AZURE, model_type="gpt-4.1", # Azure 部署名 api_key=os.getenv("AZURE_OPENAI_API_KEY"), url=os.getenv("AZURE_OPENAI_BASE_URL"), api_version="2024-12-01-preview", ) # 定义系统消息 sys_msg = "You are a helpful assistant." # 创建 Agent camel_agent = ChatAgent(system_message=sys_msg, model=model) user_msg = """Say hi to CAMEL AI, one open-source community dedicated to the study of autonomous and communicative agents.""" # 获取回复 response = camel_agent.step(user_msg) print(response.msgs[0].content)

要点:

  • model_type传的是部署名字符串(如"gpt-4.1"),也可传ModelType枚举,如ModelType.GPT_4OModelType.GPT_4_1_MINI(测试用例 test/models/test_azure_openai_model.py 对GPT_3_5_TURBOGPT_4GPT_4_TURBOGPT_4OGPT_4O_MINI等枚举均做了参数化验证);
  • 显式传入api_version是最稳妥的写法,可避免依赖环境变量;
  • 通过ChatAgentstep()即可完成一次对话推理,response.msgs[0].content为生成的文本内容。

核心方法解析与请求路由

参考文档列出了AzureOpenAIModel的几个核心方法,这些方法从OpenAIModel继承并在 Azure 客户端上复用,其职责与返回类型如下:

方法职责返回类型
token_counter返回与该模型 token 化风格一致的计数器BaseTokenCounter(默认OpenAITokenCounter
_run执行 Azure OpenAI Chat Completion 推理ChatCompletion(非流式)或Stream[ChatCompletionChunk](流式);结构化输出流式时返回ChatCompletionStreamManager[BaseModel]
_request_chat_completion底层 Chat Completions 请求(chat.completions.createChatCompletionStream[ChatCompletionChunk]
_request_parse结构化输出请求(beta.chat.completions.parseChatCompletion
_request_stream_parse流式结构化输出解析ChatCompletionStreamManager[BaseModel]
stream返回模型是否处于流式模式bool

请求路由:chat_completions 还是 responses?

_run是推理的统一入口,其路由行为取决于api_mode(相关逻辑有明确测试佐证):

  • 默认模式下调用self._client.chat.completions.create(...),且不会触碰responses.create
  • "responses"模式下调用self._client.responses.create(...),并传入model(部署名)与stream标志。

测试用例 test/models/test_azure_openai_model.py 通过 mock 客户端分别断言了两种模式只走对应的调用路径,可用test/models/test_azure_openai_model.py中相关用例验证。stream方法的布尔值则取自model_config_dict中的stream配置项,用于上层判断是否按流式协议消费结果。

Responses API 模式实战

api_mode="responses"时,AzureOpenAIModel走 Azure OpenAI Responses API。这一模式的核心优势是有状态的响应链:通过previous_response_id把多轮请求串成状态链,减少重复上下文传输。源码中通过_responses_previous_response_id_by_session_responses_last_message_count_by_session两个字典按会话(session)维护链式状态,测试断言了每次成功调用后 response ID 会被保存(test/models/test_azure_openai_model.py)。

仓库提供了完整的五合一示例 examples/models/azure_openai_responses_api_example.py,覆盖基本对话、工具调用、流式工具调用、结构化输出与流式结构化输出,其环境要求为:

export AZURE_OPENAI_API_KEY="your_azure_openai_api_key" export AZURE_OPENAI_BASE_URL="https://<resource>.openai.azure.com/" export AZURE_API_VERSION="2025-03-01-preview"

注意:model_type传入的部署名必须与 Azure 部署名完全一致。

创建 Responses 模式模型

from camel.models import ModelFactory from camel.types import ModelPlatformType, ModelType def create_responses_model( stream: bool = False, model_type: ModelType = ModelType.GPT_4_1_MINI, temperature: float = 0.2, tools: list | None = None, ): model_config = { "temperature": temperature, "stream": stream, } if tools: model_config["tools"] = tools if stream: model_config["stream_options"] = {"include_usage": True} return ModelFactory.create( model_platform=ModelPlatformType.AZURE, model_type=model_type, model_config_dict=model_config, api_mode="responses", )

场景一:工具调用

from camel.toolkits import FunctionTool from camel.agents import ChatAgent def get_weather(city: str) -> str: r"""A tiny tool function for demo purposes.""" fake_weather = { "beijing": "sunny, 28C", "new york": "cloudy, 19C", "san francisco": "foggy, 16C", } return fake_weather.get(city.lower(), f"unknown weather for {city}") weather_tool = FunctionTool(get_weather) model = create_responses_model( stream=False, model_type=ModelType.GPT_4_1, temperature=0.0, tools=[weather_tool.get_openai_tool_schema()], ) agent = ChatAgent( system_message="You are a weather assistant. Use tools when needed.", model=model, tools=[weather_tool], ) resp = agent.step("How is the weather in Beijing today?") print(resp.msgs[0].content) print("tool_calls:", resp.info.get("tool_calls"))

Responses API 模式下,底层会把 OpenAI 风格的tool_calls历史转换为function_call/function_call_output输入项——该转换逻辑(_convert_messages_to_responses_input)在测试 test/models/test_azure_openai_model.py 中有明确断言。

场景二:结构化输出

from pydantic import BaseModel, Field class TravelAdvice(BaseModel): city: str = Field(description="City name") clothing: str = Field(description="Recommended clothing") reason: str = Field(description="Brief reasoning") model = create_responses_model(stream=False) agent = ChatAgent( system_message="You are a travel assistant.", model=model, ) resp = agent.step( "I am going to New York in autumn. Give advice as JSON.", response_format=TravelAdvice, ) print("raw:", resp.msgs[0].content) print("parsed:", resp.msgs[0].parsed)

当传入response_format(Pydantic 模型)时,底层会为 Responses API 附加 JSON Schema 约束:测试断言配置中会出现{"type": "json_schema", "name": "MySchema"}(test/models/test_azure_openai_model.py),保证模型输出严格符合 schema。

场景三:流式输出与流式工具调用

流式场景需将stream=True,并配合ChatAgent(..., stream_accumulate=False)逐块消费:

stream_resp = agent.step("How is the weather in Beijing today?") full_text = "" for chunk in stream_resp: msg = chunk.msgs[0] if msg.content: full_text += msg.content print(msg.content, end="", flush=True) print() print("final:", full_text) print("tool_calls:", stream_resp.info.get("tool_calls")) print("usage:", stream_resp.info.get("usage"))

流式模式下可通过stream_options={"include_usage": True}让响应携带 usage 统计,便于做 token 计量与成本核算。

测试验证:行为保障一览

仓库通过 test/models/test_azure_openai_model.py 对AzureOpenAIModel的关键行为做了系统性验证,可作为你理解实现细节与排查问题的参考:

测试用例验证点
test_openai_model/test_openai_model_create枚举类型创建、model_config_dict透传、token_counter类型、token 限制与 tiktoken 取值
test_api_mode_default_is_chat_completions默认api_mode == "chat_completions"
test_api_mode_responses_sets_mode"responses"模式初始化链式状态字典
test_api_mode_invalid_raises非法api_mode抛出ValueError
test_normalize_tools_for_responses_api工具 schema 从嵌套function结构规范化为扁平name/parameters结构
test_convert_messages_tool_call_history历史工具调用消息转换为function_call/function_call_output
test_run_routes_to_responses_api"responses"模式下_run调用responses.create
test_run_chat_completions_mode_does_not_call_responses默认模式只调用chat.completions.create
test_responses_chain_state_saved_after_run成功调用后保存 response ID 以支持状态链
test_prepare_responses_request_config_structured_outputresponse_format转换为json_schema文本格式
test_prepare_responses_request_config_n_warningResponses 模式不支持n > 1,发出警告并移除n参数

进阶应用:Prompt Caching 与混合多智能体

Azure 自动 Prompt Caching

Azure OpenAI 对 GPT-4o 及以上模型、且请求前缀达到 1024+ token 时会自动启用缓存。仓库示例 examples/models/prompt_caching_azure_example.py 展示了在 CAMEL 中让 Agent 自动抓取并分析博客内容、配合ChatGPTConfig(prompt_cache_key=...)显式标记缓存的写法:

model = ModelFactory.create( model_platform=ModelPlatformType.AZURE, model_type=deployment_name, # 默认 gpt-4o api_version=api_version, # 默认 2024-12-01-preview model_config_dict=ChatGPTConfig( prompt_cache_key="blog_analysis_cache", # 可选,显式缓存键 ).as_dict(), ) agent = ChatAgent( system_message="You are a helpful assistant.", model=model, tools=[FunctionTool(fetch_url)], )

该示例还演示了model_type直接读取AZURE_DEPLOYMENT_NAME环境变量(缺省"gpt-4o")的兼容写法。

混合多智能体:Azure + Claude 协作

CAMEL 的模型后端彼此独立,因此可以在同一个Workforce(多智能体工作流)中混用不同厂商的模型。仓库 cookbook docs/cookbooks/multi_agent_society/azure_openai_claude_society.md 演示了如何用 Claude 4 与 Azure OpenAI 组合出多研究者协作的智能体社会(ARENA AI Alignment 研究场景),其中 Azure 模型正是通过ModelPlatformType.AZUREAzureOpenAIModel注入的。这印证了统一模型后端的设计价值:上层任务编排完全感知不到底层厂商差异。

常见问题与注意事项

  1. api_version必填:构造参数与环境变量AZURE_API_VERSION至少提供一个,否则抛出ValueError。版本号需与你的 Azure 资源支持范围匹配。
  2. model_type即部署名:传错部署名会导致 404 或认证失败;部署名不一定等于模型 ID,请以 Azure 门户中实际创建的部署为准。
  3. azure_deployment_name已弃用:当前版本会忽略该参数并发出DeprecationWarning,请迁移到model_type
  4. Responses 模式的限制n > 1不受支持,传入会告警并被移除;工具调用历史需要按function_call/function_call_output结构转换(框架已自动处理)。
  5. 自定义客户端接口约束:传入自定义client/async_client时(如 RL 框架 AReaL、rLLM 提供的兼容客户端),必须实现.chat.completions.create().beta.chat.completions.parse()接口,且此时kwargs中的客户端初始化参数会被忽略。
  6. 认证方式:除 API Key 外,支持通过azure_ad_token(Entra ID 令牌)与azure_ad_token_provider(每次请求调用的令牌提供函数,适合自动刷新)进行认证。
  7. 超时与重试timeout未指定时依次回落MODEL_TIMEOUT环境变量、默认 180 秒;max_retries默认 3 次,均可按需调整。

如需进一步深入,可继续阅读参考文档 docs/reference/camel.models.azure_openai_model.md、继承自OpenAIModel的完整方法实现 camel/models/openai_model.py、模型工厂 camel/models/model_factory.py,以及上文提到的 基础示例、Responses API 示例 与 测试用例。

【免费下载链接】camel🐫 CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询