• python版本：3.12
• pydantic-ai版本：1.70.0
• 操作系统：windows11

一、问题抛出

作为Pydantic，FastAPI，Typer的忠实用户，看到pydantic团队的新作品pydantic-ai后也是第一时间尝尝鲜了。

于是本人便写下了如下demo：

from pydantic_ai import Agent
import os


# pydantic-ai调用moonshot模型必须配置MOONSHOTAI_API_KEY环境变量
assert os.getenv('MOONSHOTAI_API_KEY') is not None


agent = Agent(
    'moonshotai:moonshot-v1-8k',
    system_prompt='请介绍一下pydantic'
)


ans = agent.run_sync()
print(ans.output)

我兴奋地查看终端想要看到预料中的输出，但是实际运行结果给我拉了坨大的：

    raise ModelHTTPError(status_code=status_code, model_name=self.model_name, body=e.body) from e
pydantic_ai.exceptions.ModelHTTPError: status_code: 401, model_name: moonshot-v1-8k, body: {'message': 'Invalid Authentication', 'type': 'invalid_authentication_error'}

这是什么意思？为什么报了401呢？

API过期了？可是用其他库写的代码都能跑通啊。

于是本人便顺着TraceBack一路打断点发现了问题所在。

首先，大部分人可能对底层源码没什么兴趣，所以本人就先果后因，先直接讲解决方案，有兴趣探究原理的可以自行查看。

二、解决方案

其实很简单，不要设置MOONSHOTAI_API_KEY这个环境变量。而是设置OPENAI_API_KEY和OPENAI_BASE_URL。如下所示：

from pydantic_ai import Agent
import os


# pydantic-ai调用openai模型必须配置OPENAI_API_KEY环境变量
# 注意：配置的是kimi的api-key
assert os.getenv('OPENAI_API_KEY') is not None

# 可以通过环境变量覆盖掉base_url
# 如果不设置这个环境变量，他就用的是openai的base_url
assert os.getenv('OPENAI_BASE_URL') == 'https://api.moonshot.cn/v1'


agent = Agent(
    'openai:moonshot-v1-8k',
    system_prompt='请介绍一下pydantic'
)


ans = agent.run_sync()
print(ans.output)

这个时候在运行，就没有问题啦。

好了，对原因没有兴趣的朋友可以改代码去了。

三、原因剖析

注：以下内容涉及pydantic-ai源码，如果您的版本不是1.70.0，可能实际内容会不一样。

首先太复杂的细节我们并不需要知道，我们只需要知道，pydantic_ai.Agent的底层其实还是openai

是封装了一个异步的openai的client

这个client可以通过调用agent._model.client获得到。

我们运行这个代码：

from pydantic_ai import Agent

agent = Agent(
    'moonshotai:moonshot-v1-8k',
    system_prompt='请介绍一下pydantic'
)

print(agent._model.client._base_url)

他的输出是：

https://api.moonshot.ai/v1/

至此真相已大白，pydantic-ai底层调用moonshot的base_url

用的是国际版的https://api.moonshot.ai/v1/

而不是国内版的https://api.moonshot.cn/v1/

由于各种原因，国内用户通常需要访问 api.moonshot.cn，而 api.moonshot.ai 是国际版端点，两者账号体系不互通。

那这个https://api.moonshot.ai/v1/是怎么来的？我要如何把他修改成https://api.moonshot.cn/v1/？

很好小子，我们就抽丝剥茧，先弄清楚agent._model这个封装属性是怎么回事。

先看Agent类的构造器，其中有这样两行：：

pydantic_ai.agent.__init__.py #L342

# defer_model_check通常为False，所以这里不用管这个if
# 直接研究`infer_model`即可
if model is None or defer_model_check:
    self._model = model
else:
    self._model = models.infer_model(model)

可以看到这个self._model是通过models.infer_model函数处理得到的。

这个models.infer_model是怎么实现的呢？这里你将看到pydantic-ai中的史诗级屎山代码：

pydantic_ai.models.__init__.py #L1202

def infer_model(  # noqa: C901
    model: Model | KnownModelName | str, provider_factory: Callable[[str], Provider[Any]] = infer_provider
) -> Model:
    """Infer the model from the name.

    Args:
        model:
            Model name to instantiate, in the format of `provider:model`. Use the string "test" to instantiate TestModel.
        provider_factory:
            Function that instantiates a provider object. The provider name is passed into the function parameter. Defaults to `provider.infer_provider`.
    """
    if isinstance(model, Model):
        return model
    elif model == 'test':
        from .test import TestModel

        return TestModel()

    provider_name, model_name = parse_model_id(model)

对于我们传入的moonshotai:moonshot-v1-8k，provider_name和model_name会分别为moonshotai和moonshotai:moonshot-v1-8k

接下来的代码是堪称地狱级别的if ... elif .. elif ...堆叠：

model_kind = provider_name
if model_kind.startswith('gateway/'):
    from ..providers.gateway import normalize_gateway_provider

    model_kind = normalize_gateway_provider(model_kind)

# OpenRouter and Cerebras need to be checked before OpenAI,
# as they are in `OpenAIChatCompatibleProvider` but have their own model classes.
if model_kind == 'openrouter':
    from .openrouter import OpenRouterModel

    return OpenRouterModel(model_name, provider=provider)
elif model_kind == 'cerebras':
    from .cerebras import CerebrasModel

    return CerebrasModel(model_name, provider=provider)
elif model_kind in ('openai-chat', 'openai', *get_args(OpenAIChatCompatibleProvider.__value__)):
    from .openai import OpenAIChatModel

    return OpenAIChatModel(model_name, provider=provider)
elif model_kind == 'openai-responses':
    from .openai import OpenAIResponsesModel

    return OpenAIResponsesModel(model_name, provider=provider)

# 这里堆叠了无数行，就不展开了

最后实际的返回值是哪一行返回的呢？

是第1252行，因为moonshot使用的也是openai协议。

那AsyncOpenAI封装在哪里呢？

查看OpenAIChatModel的构造器：

pydantic_ai.models.openai.py #L559

def __init__(
    self,
    model_name: OpenAIModelName,
    *,
    provider: OpenAIChatCompatibleProvider
    | Literal[
        'openai',
        'openai-chat',
        'gateway',
    ]
    | Provider[AsyncOpenAI] = 'openai',
    profile: ModelProfileSpec | None = None,
    system_prompt_role: OpenAISystemPromptRole | None = None,
    settings: ModelSettings | None = None,
):
    """Initialize an OpenAI model.

    Args:
        model_name: The name of the OpenAI model to use. List of model names available
            [here](https://github.com/openai/openai-python/blob/v1.54.3/src/openai/types/chat_model.py#L7)
            (Unfortunately, despite being ask to do so, OpenAI do not provide `.inv` files for their API).
        provider: The provider to use. Defaults to `'openai'`.
        profile: The model profile to use. Defaults to a profile picked by the provider based on the model name.
        system_prompt_role: The role to use for the system prompt message. If not provided, defaults to `'system'`.
            In the future, this may be inferred from the model name.
        settings: Default model settings for this model instance.
    """
    self._model_name = model_name

    if isinstance(provider, str):
        provider = infer_provider('gateway/openai' if provider == 'gateway' else provider)
    self._provider = provider
    self.client = provider.client

找到你了，原来AsyncOpenAI封装在这个provider里。

注意pydantic_ai.models.__init__.py #L1231：

provider = provider_factory(provider_name)

provide是这里传来的！

很好，我们查看一下provider_factory或者说infer_provider是如何实现的？

pydantic_ai.providers.__init__.py #L189

def infer_provider(provider: str) -> Provider[Any]:
    """Infer the provider from the provider name."""
    if provider.startswith('gateway/'):
        from .gateway import gateway_provider

        upstream_provider = provider.removeprefix('gateway/')
        return gateway_provider(upstream_provider)
    elif provider in ('google-vertex', 'google-gla', 'vertexai'):
        from .google import GoogleProvider

        return GoogleProvider(vertexai=provider in ('google-vertex', 'vertexai'))
    else:
        provider_class = infer_provider_class(provider)
        return provider_class()

天呐，又得跳到infer_provider_class里面看源码，那就跳吧。

我去……，又是疯狂的if ... elif .. elif ...地狱：

pydantic_ai.providers.__init__.py #L55

def infer_provider_class(provider: str) -> type[Provider[Any]]:  # noqa: C901
    """Infers the provider class from the provider name."""
    # Normalize gateway-prefixed providers (e.g. 'gateway/openai' -> 'openai')
    if provider.startswith('gateway/'):
        from .gateway import normalize_gateway_provider

        provider = normalize_gateway_provider(provider)

    # Normalize deprecated/alias provider names
    if provider == 'vertexai':
        provider = 'google-vertex'
    elif provider == 'google':
        provider = 'google-gla'

    if provider in ('openai', 'openai-chat', 'openai-responses'):
        from .openai import OpenAIProvider

        return OpenAIProvider
    elif provider == 'deepseek':
        from .deepseek import DeepSeekProvider

        return DeepSeekProvider
    elif provider == 'openrouter':
        from .openrouter import OpenRouterProvider

        return OpenRouterProvider
    elif provider == 'vercel':
        from .vercel import VercelProvider

        return VercelProvider
    elif provider == 'azure':
        from .azure import AzureProvider

        return AzureProvider
    elif provider in ('google-vertex', 'google-gla'):
        from .google import GoogleProvider

        return GoogleProvider
    elif provider == 'bedrock':
        from .bedrock import BedrockProvider

我们再看看具体到provider为moonshotai时，代码返回什么？

锁定到第125行：

elif provider == 'moonshotai':
    from .moonshotai import MoonshotAIProvider

    return MoonshotAIProvider

这个时候再查看MoonshotAIProvider的具体实现：

pydantic_ai.providers.moonshotai.py

class MoonshotAIProvider(Provider[AsyncOpenAI]):
    """Provider for MoonshotAI platform (Kimi models)."""

    @property
    def name(self) -> str:
        return 'moonshotai'

    @property
    def base_url(self) -> str:
        # OpenAI-compatible endpoint, see MoonshotAI docs
        return 'https://api.moonshot.ai/v1'

    @property
    def client(self) -> AsyncOpenAI:
        return self._client

真相大白，最终的问题出在这里，pydantic-ai把base_url写死了。

那如何修复问题呢？相信经常用openai的小朋友们这个时候会想到，使用openai构造client的时候，如果没有传参base_url

openai就会从环境变量读。如果没有读到，就是用使用默认的openai的base_url

这个时候查看OpenAIProvider的代码实现：

pydantic_ai.providers.openai.py

class OpenAIProvider(Provider[AsyncOpenAI]):
    """Provider for OpenAI API."""

    @property
    def name(self) -> str:
        return 'openai'

    @property
    def base_url(self) -> str:
        return str(self.client.base_url)

    @property
    def client(self) -> AsyncOpenAI:
        return self._client

他的base_url不是写死的！

那就只要让代码使用OpenAIProvider并提供相应环境变量，而不要使用MoonshotAIProvider不就行了？

现在我们回到解决方案那里给的代码：

from pydantic_ai import Agent
import os


# pydantic-ai调用openai模型必须配置OPENAI_API_KEY环境变量
# 注意：配置的是kimi的api-key
assert os.getenv('OPENAI_API_KEY') is not None

# 可以通过环境变量覆盖掉base_url
# 如果不设置这个环境变量，他就用的是openai的base_url
assert os.getenv('OPENAI_BASE_URL') == 'https://api.moonshot.cn/v1'


agent = Agent(
    'openai:moonshot-v1-8k',
    system_prompt='请介绍一下pydantic'
)


ans = agent.run_sync()
print(ans.output)

这个代码最后运行时得到的base_url就不是写死的https://api.moonshot.ai/v1/，而是我们配置的环境变量了。