旧式输入和输出代理架构

2025-10-30

注释

Databricks 建议迁移到 ResponsesAgent 架构以创作代理。请参阅代码中的作者 AI 代理。

AI 代理必须遵循特定的输入和输出架构要求，才能与 Databricks 上的其他功能兼容。本页介绍如何使用旧代理创作签名和接口： ChatAgent 接口、 ChatModel 接口、 SplitChatMessageRequest 输入架构和 StringResponse 输出架构。

创作旧 ChatAgent 代理

MLflow ChatAgent 接口类似于 OpenAI ChatCompletion 架构，但不严格兼容。

ChatAgent 可以轻松包装现有代理，实现 Databricks 兼容性。

若要了解如何创建 ChatAgent，请参阅以下部分中的示例，并 MLflow 文档 - 什么是 ChatAgent 接口。

若要使用 ChatAgent 创作和部署代理，请安装以下各项：

databricks-agents 0.16.0 或更高版本
mlflow 2.20.2 或更高版本
Python 3.10 或更高版本。
- 若要满足此要求，可以使用无服务器计算或 Databricks Runtime 13.3 LTS 或更高版本。

%pip install -U -qqqq databricks-agents>=0.16.0 mlflow>=2.20.2

如果我已有代理，该怎么办？

如果已有使用 LangChain、LangGraph 或类似框架构建的代理，则无需重写代理以在 Databricks 上使用它。只需使用 MLflow ChatAgent 接口包装现有代理即可：

编写继承自 mlflow.pyfunc.ChatAgent. 的 Python 包装类。

在包装类中，将现有代理保留为属性 self.agent = your_existing_agent。
该 ChatAgent 类要求你实现一个 predict 处理非流式处理请求的方法。

predict 必须接受：
- messages: list[ChatAgentMessage] 是一个包含多个角色的列表 ChatAgentMessage（例如“用户”或“助手”），每个角色都有一个提示和一个 ID。
- （可选）context: Optional[ChatContext] 和 custom_inputs: Optional[dict] 用于额外数据。
```
import uuid

# input example
[
  ChatAgentMessage(
    id=str(uuid.uuid4()),  # Generate a unique ID for each message
    role="user",
    content="What's the weather in Paris?"
  )
]
```
predict 必须返回一个 ChatAgentResponse。
```
import uuid

# output example
ChatAgentResponse(
  messages=[
    ChatAgentMessage(
      id=str(uuid.uuid4()),  # Generate a unique ID for each message
      role="assistant",
      content="It's sunny in Paris."
    )
  ]
)
```
在格式之间转换

在 predict中，将传入消息转换为 list[ChatAgentMessage] 代理所需的输入格式。

代理生成响应后，将其输出转换为一个或多个 ChatAgentMessage 对象，并将其包装在一个 ChatAgentResponse中。

小窍门

自动转换 LangChain 输出

如果要包装 LangChain 代理，可以使用 mlflow.langchain.output_parsers.ChatAgentOutputParser 它自动将 LangChain 输出转换为 MLflow ChatAgentMessage 和 ChatAgentResponse 架构。

下面是用于转换代理的简化模板：

from mlflow.pyfunc import ChatAgent
from mlflow.types.agent import ChatAgentMessage, ChatAgentResponse, ChatAgentChunk
import uuid


class MyWrappedAgent(ChatAgent):
  def __init__(self, agent):
    self.agent = agent

  def predict(self, messages, context=None, custom_inputs=None):
    # Convert messages to your agent's format
    agent_input = ... # build from messages
    agent_output = self.agent.invoke(agent_input)
    # Convert output to ChatAgentMessage
    return ChatAgentResponse(
      messages=[ChatAgentMessage(role="assistant", content=agent_output, id=str(uuid.uuid4()),)]
    )

  def predict_stream(self, messages, context=None, custom_inputs=None):
    # If your agent supports streaming
    for chunk in self.agent.stream(...):
      yield ChatAgentChunk(delta=ChatAgentMessage(role="assistant", content=chunk, id=str(uuid.uuid4())))

有关完整示例，请参阅以下部分中的笔记本。

`ChatAgent` 示例

以下笔记本演示如何使用常用库 OpenAI、LangGraph 和 AutoGen 编写流式和非流式处理的ChatAgents。

LangGraph

LangGraph 工具调用代理

获取笔记本

OpenAI

AutoGen

AutoGen 工具调用代理

获取笔记本

DSPy

DSPy 聊天专用代理

获取笔记本

若要了解如何通过添加工具扩展这些代理的功能，请参阅 AI 代理工具。

流式处理 ChatAgent 响应

流式处理代理以较小增量区块的连续流形式提供响应。流式处理可减少感知到的延迟并改善会话代理的用户体验。

若要创作流式 ChatAgent，需要定义一个方法，该方法返回一个生成器，该生成器会生成 predict_stream 对象——每个 ChatAgentChunk 对象都包含响应的一部分。详细了解 ChatAgent的理想流式处理行为。

以下代码演示示例 predict_stream 函数，有关流式处理代理的完整示例，请参阅 ChatAgent 示例：

def predict_stream(
  self,
  messages: list[ChatAgentMessage],
  context: Optional[ChatContext] = None,
  custom_inputs: Optional[dict[str, Any]] = None,
) -> Generator[ChatAgentChunk, None, None]:
  # Convert messages to a format suitable for your agent
  request = {"messages": self._convert_messages_to_dict(messages)}

  # Stream the response from your agent
  for event in self.agent.stream(request, stream_mode="updates"):
    for node_data in event.values():
      # Yield each chunk of the response
      yield from (
        ChatAgentChunk(**{"delta": msg}) for msg in node_data["messages"]
      )

创作旧版 ChatModel 代理

重要

Databricks 建议 ChatAgent 用于创建代理或生成式人工智能应用的接口。若要从 ChatModel 迁移到 ChatAgent，请参阅 MLflow 文档 - 从 ChatModel 迁移到 ChatAgent。

ChatModel 是 MLflow 中的旧代理创作接口，它扩展了 OpenAI 的 ChatCompletion 架构，允许你在添加自定义功能时保持与支持 ChatCompletion 标准的平台的兼容性。有关更多详细信息，请参阅 MLflow：ChatModel 入门。

将代理编写为 mlflow.pyfunc.ChatModel 的子类具有以下优势：

在调用已部署代理时，启用流式传输代理输出（在请求正文中绕过 {stream: true}）。
提供代理时自动启用 AI 网关推理表，从而提供对增强请求日志元数据（例如请求者名称）的访问权限。

警告

请求日志和评估日志已弃用，将在将来的版本中删除。有关迁移指南，请参阅请求日志和评估日志弃用。
允许使用类型化的 Python 类编写与 ChatCompletion 架构兼容的代理代码。
MLflow 在记录代理时自动推断聊天完成兼容的签名，即使没有 input_example。这简化了注册和部署代理的过程。请参阅在日志记录期间推断模型签名。

以下代码最好在 Databricks 笔记本中运行。笔记本提供了一个方便的环境，用于开发、测试和迭代代理。

MyAgent 类扩展 mlflow.pyfunc.ChatModel，实现所需的 predict 方法。这可确保与马赛克 AI 代理框架兼容。

该类还包括用于处理流输出的可选方法 _create_chat_completion_chunk 和 predict_stream。

# Install the latest version of mlflow
%pip install -U mlflow
dbutils.library.restartPython()

import re
from typing import Optional, Dict, List, Generator
from mlflow.pyfunc import ChatModel
from mlflow.types.llm import (
  # Non-streaming helper classes
  ChatCompletionRequest,
  ChatCompletionResponse,
  ChatCompletionChunk,
  ChatMessage,
  ChatChoice,
  ChatParams,
  # Helper classes for streaming agent output
  ChatChoiceDelta,
  ChatChunkChoice,
)

class MyAgent(ChatModel):
  """
  Defines a custom agent that processes ChatCompletionRequests
  and returns ChatCompletionResponses.
  """
  def predict(self, context, messages: list[ChatMessage], params: ChatParams) -> ChatCompletionResponse:
    last_user_question_text = messages[-1].content
    response_message = ChatMessage(
      role="assistant",
      content=(
        f"I will always echo back your last question. Your last question was: {last_user_question_text}. "
      )
    )
    return ChatCompletionResponse(
      choices=[ChatChoice(message=response_message)]
    )

  def _create_chat_completion_chunk(self, content) -> ChatCompletionChunk:
    """Helper for constructing a ChatCompletionChunk instance for wrapping streaming agent output"""
    return ChatCompletionChunk(
      choices=[ChatChunkChoice(
        delta=ChatChoiceDelta(
          role="assistant",
          content=content
        )
      )]
    )

  def predict_stream(
    self, context, messages: List[ChatMessage], params: ChatParams
  ) -> Generator[ChatCompletionChunk, None, None]:
    last_user_question_text = messages[-1].content
    yield self._create_chat_completion_chunk(f"Echoing back your last question, word by word.")
    for word in re.findall(r"\S+\s*", last_user_question_text):
      yield self._create_chat_completion_chunk(word)

agent = MyAgent()
model_input = ChatCompletionRequest(
  messages=[ChatMessage(role="user", content="What is Databricks?")]
)
response = agent.predict(context=None, messages=model_input.messages, params=None)
print(response)

在一个笔记本中定义代理类 MyAgent 时，建议创建单独的驱动程序笔记本。驱动程序笔记本将代理记录到模型注册表，并使用 Model Service 部署代理。

此分离遵循 Databricks 推荐的工作流，使用 MLflow 的 "Models from Code" 方法来记录模型。

SplitChatMessageRequest 输入架构（已弃用）

SplitChatMessagesRequest 允许将当前查询和历史记录单独作为代理输入传递。

  question = {
    "query": "What is MLflow",
    "history": [
      {
        "role": "user",
        "content": "What is Retrieval-augmented Generation?"
      },
      {
        "role": "assistant",
        "content": "RAG is"
      }
    ]
  }

StringResponse 输出架构（已弃用）

StringResponse 允许以具有单个字符串 content 字段的对象的形式返回代理的响应：

{"content": "This is an example string response"}

反馈

此页面是否有帮助？

通过

旧式输入和输出代理架构

创作旧 ChatAgent 代理

如果我已有代理，该怎么办？

ChatAgent 示例

LangGraph

LangGraph 工具调用代理

OpenAI

OpenAI 工具调用代理

OpenAI 响应 API 工具调用代理

仅限聊天的 OpenAI 代理

AutoGen

AutoGen 工具调用代理

DSPy

DSPy 聊天专用代理

流式处理 ChatAgent 响应

创作旧版 ChatModel 代理

SplitChatMessageRequest 输入架构 （已弃用）

StringResponse 输出架构（已弃用）

反馈

其他资源

`ChatAgent` 示例

SplitChatMessageRequest 输入架构（已弃用）