实战指南：AgentScope 自定义模型集成，一次讲清 ChatModelBase 契约与完整代码

实战指南：AgentScope 自定义模型集成，一次讲清 ChatModelBase 契约与完整代码

【免费下载链接】agentscopeBuild and run agents you can see, understand and trust.项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope

把公司内部 AI 服务接进 AgentScope，是让私有模型获得工具调用、上下文压缩等框架能力的常见需求。这篇实战指南按真实代码结构，带你从零完成一次自定义模型集成：读懂基类契约、写好模型文件、注册导出，再用几行 smoke test 确认整条链路跑通。

ChatModelBase 契约：你必须实现的三样东西

AgentScope 里所有对话模型类都继承自ChatModelBase（src/agentscope/model/_base.py）。它是模型层的统一契约，上游的 Agent、Agent Service 只认这个抽象，不关心背后接的是哪家服务。

继承它需要遵守三件事：

• 初始化参数：credential（继承CredentialBase的凭据对象，用来挂 API 密钥等敏感信息）、model（模型名字符串）、parameters（参数配置），以及stream这类行为开关，都要透传给父类构造。
• 核心方法：公开入口__call__基类已经写好（含重试与流式包装），你要实现的是它的下层抽象方法_call_api——真正发请求、解析响应的地方。
• 返回值：非流式返回一个ChatResponse；流式则返回由ChatResponse组成的异步生成器（一块块增量，最后一块标记结束）。

契约示意（关键签名，不超过 8 行）：

class MyChatModel(ChatModelBase): async def __call__(self, messages, tools=None, tool_choice=None, **kwargs): ... # 基类实现：重试 + 流式包装 @abstractmethod async def _call_api(self, model_name, messages, tools=None, tool_choice=None, **kwargs): """必须实现，返回 ChatResponse | AsyncGenerator"""

工具调用场景下，基类还提供了_validate_tool_choice校验方法，用来确认tool_choice指向的工具确实存在，避免模型"幻觉"出不存在的工具名。

从零写一个模型文件

下面这套代码是放在你本地项目里新建的文件，仓库本身保持只读不动。

先建文件，比如my_model/my_chat_model.py，并定义凭据类。凭据类是敏感信息的唯一入口，密钥不进模型代码，这一步省掉后患：

from agentscope.credential import CredentialBase from agentscope.model import ChatModelBase from agentscope.model._model_response import ChatResponse from pydantic import Field class MyCredential(CredentialBase): """内部 AI 服务的访问凭据""" token: str = Field(default="", description="Access token") @classmethod def get_chat_model_class(cls): from my_model.my_chat_model import MyChatModel return MyChatModel

再实现模型类。__init__必须把参数原样透传给父类，否则基类里stream、max_retries这些机制会拿不到配置：

import httpx from datetime import datetime from agentscope.message import Msg, TextBlock class MyChatModel(ChatModelBase): """对接内部 AI 服务的自定义模型""" class Parameters(BaseModel): max_tokens: int | None = None def __init__(self, credential, model, parameters=None, stream=True, **kwargs): super().__init__( credential=credential, model=model, parameters=parameters or self.Parameters(), stream=stream, **kwargs, ) self._http = httpx.AsyncClient( base_url=credential.token, # 简化示意 timeout=60.0, )

然后是_call_api。这里是自定义模型集成的核心：把框架的Msg列表转成你的服务能懂的请求，再把响应翻译回ChatResponse。流式分支不能省——Agent 的实时输出全靠它：

async def _call_api(self, model_name, messages, tools=None, tool_choice=None, **kwargs): if tool_choice and tools: self._validate_tool_choice(tool_choice, tools) payload = { "model": model_name, "messages": [m.get_content_blocks() for m in messages], "stream": self.stream, } if not self.stream: return await self._parse_completion(payload) # 流式：逐块 yield，最后一块 is_last=True async def _stream(): async with self._http.stream( "POST", "/v1/chat", json=payload, ) as resp: async for line in resp.aiter_lines(): chunk = self._parse_chunk(line) yield chunk # is_last=False 的增量块 return _stream() async def _parse_completion(self, payload) -> ChatResponse: res = await self._http.post("/v1/chat", json=payload) data = res.json() return ChatResponse( content=[TextBlock(text=data["message"]["content"])], is_last=True, ) def _parse_chunk(self, line) -> ChatResponse: data = json.loads(line.removeprefix("data: ")) last = data.get("finish", False) return ChatResponse( content=[TextBlock(text=data["delta"])] if not last else [], is_last=last, )

两个容易忽略的细节：增量块统一is_last=False，只有终止块置True，基类靠它触发完整响应的收尾；另外基类__call__会自动重试你通过_get_retryable_exceptions声明的异常类型，网络层瞬时错误不用自己处理。

模型注册导出 + 5 行 smoke test

类写好后，最后一步是导出。这一步的意义在于：agentscope.model包名空间的__all__决定了from agentscope.model import MyChatModel这类写法是否可用——漏了它，下游项目会直接 ImportError。在你的本地包__init__.py中加两行即可（官方各模型在 src/agentscope/model/init.py 中的注册方式就是参照样板）：

from .my_chat_model import MyChatModel __all__ = [..., "MyChatModel"]

导出完成后，别急着上功能，先跑最小闭环。下面这个 smoke test 不超过 5 行核心断言，验证的正是集成最脆弱的那一环——消息进去、ChatResponse出来：

import asyncio async def main(): m = MyChatModel(MyCredential(token="http://127.0.0.1:8080"), model="internal-llm", stream=False) res = await m([Msg("user", [TextBlock(text="ping")], "user")]) assert res.is_last and res.content asyncio.run(main())

返回了带内容的is_last=True响应，说明导入、实例化、请求解析整条链路都通了。后续再开流式、加工具，出问题时定位范围会小很多。

踩坑速查表

上生产前补三件事

• 密钥走凭据对象：API 密钥只放进CredentialBase子类并从环境变量读取，示例代码里写死 token 仅限本地调试。
• 重试靠基类：覆盖_get_retryable_exceptions声明哪些异常值得重试，基类会按max_retries/retry_delay自动处理，不用手写 sleep 循环。
• 补全用量数据：在ChatResponse.usage里填 token 数与耗时，配合追踪中间件（src/agentscope/middleware/_tracing/）记录调用指标，线上排查才有据可依。

建议的下一步很具体：先在本地项目建一个最简模型，只跑非流式文本，用上面的 smoke test 验证闭环；跑稳之后再打开stream分支，最后才叠加工具调用与结构化输出。一次加一块，每块都先验证——这是流式模型接入里最省时间的做法。

【免费下载链接】agentscopeBuild and run agents you can see, understand and trust.项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope