要用Python调通Mage多模态对话，核心是先定好HTTP请求骨架，而不是依赖特定SDK。以下代码假设Mage服务暴露/chat端点，接收含文本和图像的JSON消息，返回带reply字段的响应。你可以根据实际接口调整URL、字段名和鉴权方式。

对于Mage这类多模态对话服务，调用骨架可以统一为：组装输入→发POST请求→解析响应。适用场景是开发者已有服务地址和接口文档；操作动作为按示例封装请求和解析；验证方式为用mock测试模拟响应；风险边界是不同部署版本字段可能不同，需要结合环境确认。

搭建一个最简的 HTTP 客户端

请求层只做一件事：把消息列表序列化成JSON，POST到Mage的接口，并返回反序列化后的字典。以下函数用requests实现了这个动作，并显式设置超时，避免连接挂死。

import requests

def call_mage_api(messages, api_url='http://127.0.0.1:8000/chat', timeout=30):
    payload = {'messages': messages}
    resp = requests.post(api_url, json=payload, timeout=timeout)
    resp.raise_for_status()  # 非2xx状态码会抛异常
    return resp.json()

这个函数是后续所有调用链路的底层。如果你的Mage服务需要鉴权，可以在headers里加Authorization字段，但这里先保持最简。验证方式：在本地启动服务后，直接调用这个函数，检查返回的dict是否符合预期。

定义多模态输入数据模型

为了不让文本、图片混杂在裸字典里，建议用dataclass固定消息结构。下面分别定义文本块和图像块，再组合成一条用户消息。

from dataclasses import dataclass, field
from typing import List, Dict

@dataclass
class ImagePart:
    type: str = 'image'
    source: str = ''  # 图片URL或base64字符串

@dataclass
class TextPart:
    type: str = 'text'
    text: str = ''

@dataclass
class Message:
    role: str = 'user'
    content: List[Dict] = field(default_factory=list)

    def add_text(self, text: str):
        self.content.append(TextPart(text=text).__dict__)

    def add_image(self, source: str):
        self.content.append(ImagePart(source=source).__dict__)

    def to_dict(self):
        return {'role': self.role, 'content': self.content}

这里用__dict__把dataclass实例转成dict。字段名需要与Mage接口约定一致；如果不一致，就改这里的属性名。如果后续新增音频等模态，只需增加对应的Part类并添加到content列表。

封装对话管理与返回解析

容器类MageClient负责维护对话历史，并把用户输入转换为消息模型，最后解析返回。假设接口返回格式为{'reply': '模型生成文本'}，你按实际情况调整取值。

class MageClient:
    def __init__(self, api_url='http://127.0.0.1:8000/chat', timeout=30):
        self.api_url = api_url
        self.timeout = timeout
        self.history = []  # 保存对话上下文

    def chat(self, text=None, image_source=None):
        msg = Message(role='user')
        if text:
            msg.add_text(text)
        if image_source:
            msg.add_image(image_source)
        if not msg.content:
            raise ValueError('至少需要提供文本或图像之一')

        messages = [m.to_dict() for m in self.history] + [msg.to_dict()]
        data = call_mage_api(messages, self.api_url, self.timeout)
        reply = data.get('reply', '')
        if not reply:
            raise RuntimeError('响应中未找到reply字段')

        self.history.append(msg)
        self.history.append(Message(role='assistant', content=[TextPart(text=reply).__dict__]))
        return reply

这里把history存成Message对象，发送前统一转dict。返回时只提取reply，其他字段如思考过程、工具调用等可自行扩展。对话管理的关键是每次请求携带完整上下文，否则模型会丢失前文。

增加超时重试与日志记录

网络调用需要具备重试和可观测性。下面定义了一个重试装饰器，对requests.RequestException进行指数退避重试，并用logging记录每次失败的上下文。

import logging
import time
from functools import wraps

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('mage_client')

def retry_on_exception(retries=3, delay=1.0, exceptions=(requests.RequestException,)):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(retries):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    logger.warning('Mage调用失败（第%d次）：%s', attempt + 1, e)
                    if attempt == retries - 1:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator

@retry_on_exception(retries=3)
def call_mage_api_with_retry(messages, api_url, timeout):
    return call_mage_api(messages, api_url, timeout)

在第三小节的chat方法中，把调用call_mage_api的那一行替换为call_mage_api_with_retry即可。重试次数和延迟可以根据网络状况调整；注意幂等性，如果请求可能产生副作用，重试需谨慎。

编写测试用例验证调用链路

用unittest配合unittest.mock模拟requests.post，这样不依赖真实服务也能验证请求构造和返回解析。下面给出两个测试：一个只传文本，一个验证图像字段是否被正确装入payload。

import unittest
from unittest.mock import patch, MagicMock

class TestMageClient(unittest.TestCase):
    @patch('requests.post')
    def test_chat_returns_reply(self, mock_post):
        mock_post.return_value = MagicMock(
            json=lambda: {'reply': '你好'},
            raise_for_status=lambda: None,
        )
        client = MageClient()
        reply = client.chat(text='你好')
        self.assertEqual(reply, '你好')

    @patch('requests.post')
    def test_chat_payload_includes_image(self, mock_post):
        def fake_json():
            return {'reply': '看到图片了'}
        mock_resp = MagicMock(json=fake_json, raise_for_status=lambda: None)
        mock_post.return_value = mock_resp

        client = MageClient()
        client.chat(text='描述图片', image_source='https://example.com/a.png')

        args, kwargs = mock_post.call_args
        payload = kwargs['json']
        content = payload['messages'][0]['content']
        self.assertTrue(any(item['type'] == 'image' for item in content))

测试中的call_args可以从mock中取出最近一次调用的参数，用于断言请求体内的结构。运行python -m unittest即可执行。注意测试文件需要能导入MageClient等定义。