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接口,使上层ChatAgent、Workforce、Task等组件无需关心底层是 OpenAI 直连还是 Azure 部署。
从源码看(camel/models/azure_openai_model.py),该类直接继承自OpenAIModel:
Inherits all Responses API, chat completion, and streaming methods from
OpenAIModel. Only the client initialization and Azure-specific configuration are different.
这意味着OpenAIModel提供的 Responses API、Chat Completion、流式推理、结构化输出等全部能力都被继承复用,AzureOpenAIModel只需覆盖两块差异化逻辑:
- 客户端初始化:使用
openai.AzureOpenAI/openai.AsyncAzureOpenAI,并注入 Azure 特有的azure_endpoint、azure_deployment、api_version、azure_ad_token等参数; - Azure 特有配置:如
api_mode("chat_completions"或"responses")、弃用参数兼容逻辑等。
因此,在 CAMEL 中通过ModelFactory创建模型时,只需要指定平台类型为ModelPlatformType.AZURE,工厂内部就会自动映射到本类。camel/models/model_factory.py 中注册了ModelPlatformType.AZURE: AzureOpenAIModel的对应关系,这是官方推荐的创建入口(下文详述)。
部署前提与环境变量
使用 Azure OpenAI 前,你需要在 Azure 门户中完成模型部署并获取三类信息:资源终结点(endpoint)、API Key与API 版本号。CAMEL 通过环境变量与构造参数双重途径读取这些配置,对应关系如下表:
| 环境变量 | 对应构造参数 | 说明 |
|---|---|---|
AZURE_OPENAI_API_KEY | api_key | Azure OpenAI 资源的 API Key,必填 |
AZURE_OPENAI_BASE_URL | url | 资源终结点,形如https://<resource>.openai.azure.com/ |
AZURE_API_VERSION | api_version | API 版本号,必填(详见下文校验逻辑),如2024-12-01-preview |
AZURE_AD_TOKEN | azure_ad_token | Azure Active Directory 令牌,使用 Entra ID 认证时设置 |
MODEL_TIMEOUT | timeout | 请求超时秒数,默认180 |
AZURE_DEPLOYMENT_NAME | model_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_type | Union[ModelType, str] | 必填 | 要创建后端的模型,必须是你在 Azure 中部署模型时选择的部署名(deployment name),而非模型 ID。可传ModelType枚举(如ModelType.GPT_4O)或字符串(如"gpt-4.1") |
model_config_dict | Optional[Dict[str, Any]] | None | 会透传给openai.ChatCompletion.create()的配置字典;为None时使用ChatGPTConfig().as_dict() |
api_key | Optional[str] | None | 认证用的 API Key,缺省回落到AZURE_OPENAI_API_KEY环境变量 |
url | Optional[str] | None | 服务地址,缺省回落到AZURE_OPENAI_BASE_URL环境变量 |
timeout | Optional[float] | None | API 调用超时(秒)。缺省时读取MODEL_TIMEOUT环境变量,最终默认180秒 |
token_counter | Optional[BaseTokenCounter] | None | 自定义 token 计数器;不提供时使用OpenAITokenCounter |
api_version | Optional[str] | None | Azure API 版本,缺省读AZURE_API_VERSION,仍为空则抛ValueError |
azure_ad_token | Optional[str] | None | Azure Active Directory 令牌(Entra ID 认证场景),缺省读AZURE_AD_TOKEN |
azure_ad_token_provider | Optional[AzureADTokenProvider] | None | 一个返回 AD 令牌的函数,每次请求都会调用;适合令牌自动刷新的场景 |
max_retries | int | 3 | API 调用的最大重试次数 |
client | Optional[Any] | None | 自定义同步AzureOpenAI客户端实例,提供后不再自建。要求实现.chat.completions.create()与.beta.chat.completions.parse()接口;典型场景是 AReaL、rLLM 等 RL 框架提供的 Azure OpenAI 兼容客户端 |
async_client | Optional[Any] | None | 自定义异步AsyncAzureOpenAI客户端实例,语义同上 |
azure_deployment_name | Optional[str] | None | 已弃用,仅为向后兼容保留,未来版本将移除;使用时会发出DeprecationWarning且参数被忽略 |
**kwargs | Any | — | 透传给客户端初始化的额外参数;提供自定义客户端时被忽略 |
参考文档还额外说明了继承而来的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):
- 检查并警告弃用的
azure_deployment_name参数与AZURE_DEPLOYMENT_NAME环境变量; - 校验
api_mode合法性,并初始化 Responses API 链式状态字典(_responses_previous_response_id_by_session、_responses_last_message_count_by_session); model_config_dict为None时回退到ChatGPTConfig().as_dict();- 依次从参数/环境变量解析
api_key、url、timeout; - 跳过
OpenAIModel.__init__(它要求OPENAI_API_KEY),直接调用BaseModelBackend.__init__完成公共初始化; - 解析
api_version并强制校验; - 若已安装 langfuse 且未传自定义客户端,会优先使用
langfuse.openai.AzureOpenAI/AsyncAzureOpenAI构造可观测客户端(实现自动链路追踪);否则构造原生AzureOpenAI/AsyncAzureOpenAI客户端,azure_endpoint取url,azure_deployment取model_type,并透传api_version、api_key、AD 令牌、timeout、max_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_4O、ModelType.GPT_4_1_MINI(测试用例 test/models/test_azure_openai_model.py 对GPT_3_5_TURBO、GPT_4、GPT_4_TURBO、GPT_4O、GPT_4O_MINI等枚举均做了参数化验证);- 显式传入
api_version是最稳妥的写法,可避免依赖环境变量; - 通过
ChatAgent的step()即可完成一次对话推理,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.create) | ChatCompletion或Stream[ChatCompletionChunk] |
_request_parse | 结构化输出请求(beta.chat.completions.parse) | ChatCompletion |
_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_output | response_format转换为json_schema文本格式 |
test_prepare_responses_request_config_n_warning | Responses 模式不支持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.AZURE与AzureOpenAIModel注入的。这印证了统一模型后端的设计价值:上层任务编排完全感知不到底层厂商差异。
常见问题与注意事项
api_version必填:构造参数与环境变量AZURE_API_VERSION至少提供一个,否则抛出ValueError。版本号需与你的 Azure 资源支持范围匹配。model_type即部署名:传错部署名会导致 404 或认证失败;部署名不一定等于模型 ID,请以 Azure 门户中实际创建的部署为准。azure_deployment_name已弃用:当前版本会忽略该参数并发出DeprecationWarning,请迁移到model_type。- Responses 模式的限制:
n > 1不受支持,传入会告警并被移除;工具调用历史需要按function_call/function_call_output结构转换(框架已自动处理)。 - 自定义客户端接口约束:传入自定义
client/async_client时(如 RL 框架 AReaL、rLLM 提供的兼容客户端),必须实现.chat.completions.create()与.beta.chat.completions.parse()接口,且此时kwargs中的客户端初始化参数会被忽略。 - 认证方式:除 API Key 外,支持通过
azure_ad_token(Entra ID 令牌)与azure_ad_token_provider(每次请求调用的令牌提供函数,适合自动刷新)进行认证。 - 超时与重试:
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),仅供参考