jianmu.message

适用对象：Agent 应用开发者 / Provider 适配者 / 核心维护者 是否必读：是 相关模块：jianmu.model, jianmu.memory

1. 模块职责

jianmu.message 定义消息领域模型、消息存储协议、消息归一化与保留策略。

如果你要理解 jianmu 里的“对话消息”如何流动，这个模块是关键入口之一。

2. 适合查什么

• 基础消息类型：Message
• 快捷构造器：system()、human()、ai()、tool()
• 消息存储协议：MessageStoreProtocol、AsyncMessageStoreProtocol
• 消息归一化：StrictMessageCoercer、LenientMessageCoercer
• 存储实现：StateMessageStore

3. 使用建议

• 普通业务代码优先通过快捷构造器产生消息
• 想定制消息兼容层时，查看 coercer 与 retention 相关对象
• Provider 编码层不要直接耦合底层存储实现

4. 最小示例

from pydantic import BaseModel
from jianmu import StateManager
from jianmu.message import StateMessageStore, human, ai

class AgentState(BaseModel):
    messages: list = []

state_manager = StateManager(AgentState)
state_manager.initialize()

store = StateMessageStore(state_manager, key="messages", max_messages=50)
store.append(human("hello"))
store.append(ai("hi"))

recent = store.get_messages(limit=10)

5. 常见入口

• 想构造消息：看 system()、human()、ai()、tool()
• 想把消息历史放进状态：看 StateMessageStore
• 想兼容宽松输入：看 LenientMessageCoercer
• 想限制保留条数：看 MaxCountRetention

6. 注意事项

• 普通应用代码通常不需要自己实现 MessageStoreProtocol
• provider 编码层应依赖 Message 领域模型，而不是直接耦合具体存储实现
• 需要外部持久化时，再实现 AsyncMessageStoreProtocol 或自定义 store

7. API 参考

message

Agent messaging protocols, message store interfaces, and retention policies in Jianmu.

Message

Bases: BaseModel

Canonical message container for conversation history and tool payloads.

Supports system, user, assistant, and tool roles. content may be plain text or a structured list of multimodal blocks.

Message is Jianmu's canonical chat/history format and is the type used across model providers, message stores, and agent state.

属性：

名称	类型	描述
id	str	Stable message identifier used for diffing, replay, and event tracking.
role	Literal['system', 'user', 'assistant', 'tool']	Chat role, one of system, user, assistant, or tool.
content	Union[str, List[Any]]	Message body as plain text or a structured block list.
name	Optional[str]	Optional speaker or tool-facing name.
tool	Optional[str]	Optional tool name associated with the message.
tool_call_id	Optional[str]	Optional provider-specific tool-call correlation ID.
tool_calls	Optional[List[Dict[str, Any]]]	Optional tool-call payloads emitted by an assistant message.
metadata	Dict[str, Any]	Arbitrary structured metadata attached to the message.

to_text

to_text(self_or_content: Any = None) -> str

Convert a Message (or arbitrary content via class-call) to plain text.

参数：

名称	类型	描述	默认
self_or_content	Any	Either a Message instance or any raw content value to convert to text.	None

返回：

类型	描述
str	Decoded string representation of the content.

源代码位于： jianmu/message/base.py

def to_text(self_or_content: Any = None) -> str:
    """Convert a Message (or arbitrary content via class-call) to plain text.

    Args:
        self_or_content: Either a ``Message`` instance or any raw content
            value to convert to text.

    Returns:
        Decoded string representation of the content.
    """
    if isinstance(self_or_content, Message):
        content = self_or_content.content
    else:
        content = self_or_content

    if content is None:
        return ""
    if isinstance(content, str):
        return content
    if isinstance(content, bytes):
        return content.decode("utf-8", "replace")
    if isinstance(content, dict):
        for key in ("text", "content", "data", "value"):
            value = content.get(key)
            if isinstance(value, str):
                return value
        return str(content)
    if isinstance(content, list):
        parts = [Message.to_text(item) for item in content]
        return "\n".join([p for p in parts if p])
    return str(content)

to_dict

to_dict() -> Dict[str, Any]

Convert the message into a plain dictionary payload.

返回：

类型	描述
Dict[str, Any]	Dictionary with id, role, and content as required fields
Dict[str, Any]	plus any optional fields that are set.

源代码位于： jianmu/message/base.py

def to_dict(self) -> Dict[str, Any]:
    """Convert the message into a plain dictionary payload.

    Returns:
        Dictionary with ``id``, ``role``, and ``content`` as required fields
        plus any optional fields that are set.
    """
    data = {
        "id": self.id,
        "role": self.role,
        "content": self.content,
    }
    if self.name:
        data["name"] = self.name
    if self.tool:
        data["tool"] = self.tool
    if self.tool_call_id:
        data["tool_call_id"] = self.tool_call_id
    if self.tool_calls:
        data["tool_calls"] = self.tool_calls
    if self.metadata:
        data["metadata"] = dict(self.metadata)
    return data

MessageEvent

Bases: BaseModel

Represents a lifecycle event of a message in the store.

属性：

名称	类型	描述
event_type	Literal['added']	Event name describing the message lifecycle transition.
message	Message	Message payload associated with the event.

MessageStoreProtocol

Bases: Protocol

Protocol for storing and retrieving message history.

Most users will interact with StateMessageStore rather than implement this directly. Custom implementations are useful for Redis, SQL, or other external backends.

append

append(message: Any) -> None

Append a single message payload.

参数：

名称	类型	描述	默认
message	Any	Message or coercible payload to append.	必需

源代码位于： jianmu/message/base.py

def append(self, message: Any) -> None:
    """Append a single message payload.

    Args:
        message: Message or coercible payload to append.
    """
    ...

append_many

append_many(messages: Sequence[Any]) -> None

Append multiple message payloads in order.

参数：

名称	类型	描述	默认
messages	Sequence[Any]	Iterable of messages or coercible payloads.	必需

源代码位于： jianmu/message/base.py

def append_many(self, messages: Sequence[Any]) -> None:
    """Append multiple message payloads in order.

    Args:
        messages: Iterable of messages or coercible payloads.
    """
    ...

get_messages

get_messages(limit: int | None = None) -> List[Message]

Return stored messages, optionally limited to the newest entries.

参数：

名称	类型	描述	默认
limit	int | None	Maximum number of messages to return from the tail of the store. Returns all messages when None.	None

返回：

类型	描述
List[Message]	Ordered list of Message objects.

源代码位于： jianmu/message/base.py

def get_messages(self, limit: int | None = None) -> List[Message]:
    """Return stored messages, optionally limited to the newest entries.

    Args:
        limit: Maximum number of messages to return from the tail of the
            store. Returns all messages when ``None``.

    Returns:
        Ordered list of ``Message`` objects.
    """
    ...

subscribe

subscribe(
    callback: Callable[[MessageEvent], None],
) -> Callable[[MessageEvent], None]

Register a callback for message lifecycle events.

参数：

名称	类型	描述	默认
callback	Callable[[MessageEvent], None]	Consumer invoked with a MessageEvent on each append.	必需

返回：

类型	描述
Callable[[MessageEvent], None]	The same callback, for use as an unsubscribe token.

源代码位于： jianmu/message/base.py

def subscribe(self, callback: Callable[[MessageEvent], None]) -> Callable[[MessageEvent], None]:
    """Register a callback for message lifecycle events.

    Args:
        callback: Consumer invoked with a ``MessageEvent`` on each append.

    Returns:
        The same callback, for use as an unsubscribe token.
    """
    ...

unsubscribe

unsubscribe(
    callback: Callable[[MessageEvent], None],
) -> None

Remove a previously registered message event callback.

参数：

名称	类型	描述	默认
callback	Callable[[MessageEvent], None]	The callback reference previously passed to subscribe.	必需

源代码位于： jianmu/message/base.py

def unsubscribe(self, callback: Callable[[MessageEvent], None]) -> None:
    """Remove a previously registered message event callback.

    Args:
        callback: The callback reference previously passed to ``subscribe``.
    """
    ...

AsyncMessageStoreProtocol

Bases: Protocol

Protocol for async message store implementations.

append async

append(message: Any) -> None

Append a single message payload asynchronously.

参数：

名称	类型	描述	默认
message	Any	Message or coercible payload to append.	必需

源代码位于： jianmu/message/base.py

async def append(self, message: Any) -> None:
    """Append a single message payload asynchronously.

    Args:
        message: Message or coercible payload to append.
    """
    ...

append_many async

append_many(messages: Sequence[Any]) -> None

Append multiple messages asynchronously.

参数：

名称	类型	描述	默认
messages	Sequence[Any]	Sequence of message payloads.	必需

源代码位于： jianmu/message/base.py

async def append_many(self, messages: Sequence[Any]) -> None:
    """Append multiple messages asynchronously.

    Args:
        messages: Sequence of message payloads.
    """
    ...

get_messages async

get_messages(limit: int | None = None) -> List[Message]

Return stored messages asynchronously.

参数：

名称	类型	描述	默认
limit	int | None	Maximum number of messages to return.	None

返回：

类型	描述
List[Message]	List of Message objects.

源代码位于： jianmu/message/base.py

async def get_messages(self, limit: int | None = None) -> List[Message]:
    """Return stored messages asynchronously.

    Args:
        limit: Maximum number of messages to return.

    Returns:
        List of Message objects.
    """
    ...

subscribe

subscribe(
    callback: Callable[[MessageEvent], None],
) -> Callable[[MessageEvent], None]

Register a callback for message lifecycle events.

参数：

名称	类型	描述	默认
callback	Callable[[MessageEvent], None]	Consumer callback for MessageEvent.	必需

返回：

类型	描述
Callable[[MessageEvent], None]	The registered callback.

源代码位于： jianmu/message/base.py

def subscribe(self, callback: Callable[[MessageEvent], None]) -> Callable[[MessageEvent], None]:
    """Register a callback for message lifecycle events.

    Args:
        callback: Consumer callback for MessageEvent.

    Returns:
        The registered callback.
    """
    ...

unsubscribe

unsubscribe(
    callback: Callable[[MessageEvent], None],
) -> None

Remove a previously registered message event callback.

参数：

名称	类型	描述	默认
callback	Callable[[MessageEvent], None]	The callback to unsubscribe.	必需

源代码位于： jianmu/message/base.py

def unsubscribe(self, callback: Callable[[MessageEvent], None]) -> None:
    """Remove a previously registered message event callback.

    Args:
        callback: The callback to unsubscribe.
    """
    ...

MessageCoercerProtocol

Bases: Protocol

Convert external payloads into Message objects.

coerce

coerce(payload: Any) -> Message | None

Convert an arbitrary payload into a Message or None.

参数：

名称	类型	描述	默认
payload	Any	Payload to convert.	必需

返回：

类型	描述
Message | None	Coerced Message object, or None if invalid.

源代码位于： jianmu/message/base.py

def coerce(self, payload: Any) -> Message | None:
    """Convert an arbitrary payload into a ``Message`` or ``None``.

    Args:
        payload: Payload to convert.

    Returns:
        Coerced Message object, or None if invalid.
    """
    ...

RetentionPolicyProtocol

Bases: Protocol

Control how message history is retained on append.

apply

apply(
    existing: List[Message], incoming: List[Message]
) -> List[Message]

Merge retained history with incoming messages.

参数：

名称	类型	描述	默认
existing	List[Message]	Existing list of messages.	必需
incoming	List[Message]	New incoming list of messages.	必需

返回：

类型	描述
List[Message]	Retained messages.

源代码位于： jianmu/message/base.py

def apply(self, existing: List[Message], incoming: List[Message]) -> List[Message]:
    """Merge retained history with incoming messages.

    Args:
        existing: Existing list of messages.
        incoming: New incoming list of messages.

    Returns:
        Retained messages.
    """
    ...

MergeFnProtocol

Bases: Protocol

Pure function protocol for combining existing history with incoming messages.

StrictMessageCoercer

Strict coercer: accepts Message or valid dict, rejects everything else.

coerce

coerce(payload: Any) -> Message | None

Coerce only valid Message instances or message dicts.

参数：

名称	类型	描述	默认
payload	Any	The message instance or dict to coerce.	必需

返回：

类型	描述
Message | None	Coerced Message instance or None.

引发：

类型	描述
ValueError	If the payload dictionary is invalid or the type is incorrect.

源代码位于： jianmu/message/coercer.py

def coerce(self, payload: Any) -> Message | None:
    """Coerce only valid ``Message`` instances or message dicts.

    Args:
        payload: The message instance or dict to coerce.

    Returns:
        Coerced Message instance or None.

    Raises:
        ValueError: If the payload dictionary is invalid or the type is incorrect.
    """
    if payload is None:
        return None
    if isinstance(payload, Message):
        return payload
    if isinstance(payload, dict):
        try:
            return Message(**payload)
        except ValidationError as e:
            raise ValueError(f"Invalid message payload dict: {e}") from e
    raise ValueError(f"Invalid message payload type: {type(payload).__name__}")

LenientMessageCoercer

Lenient coercer for migration/debug use.

coerce

coerce(payload: Any) -> Message | None

Best-effort coerce arbitrary payloads into Message objects.

参数：

名称	类型	描述	默认
payload	Any	The payload to coerce.	必需

返回：

类型	描述
Message | None	Coerced Message instance or None.

源代码位于： jianmu/message/coercer.py

def coerce(self, payload: Any) -> Message | None:
    """Best-effort coerce arbitrary payloads into ``Message`` objects.

    Args:
        payload: The payload to coerce.

    Returns:
        Coerced Message instance or None.
    """
    if payload is None:
        return None
    if isinstance(payload, Message):
        return payload
    if isinstance(payload, dict):
        try:
            return Message(**payload)
        except Exception:
            return Message(role="assistant", content=str(payload))
    return Message(role="assistant", content=str(payload))

MaxCountRetention

MaxCountRetention(max_messages: Optional[int] = 50)

Retention policy that keeps only the newest N messages.

属性：

名称	类型	描述
max_messages		Maximum number of messages retained after each append.

Create a max-count retention policy.

参数：

名称	类型	描述	默认
max_messages	Optional[int]	Maximum number of messages to retain.	50

源代码位于： jianmu/message/retention.py

def __init__(self, max_messages: Optional[int] = 50):
    """Create a max-count retention policy.

    Args:
        max_messages: Maximum number of messages to retain.
    """
    self.max_messages = max_messages

apply

apply(
    existing: List[Message], incoming: List[Message]
) -> List[Message]

Append incoming messages and truncate to the newest entries.

参数：

名称	类型	描述	默认
existing	List[Message]	List of existing messages.	必需
incoming	List[Message]	List of new incoming messages.	必需

返回：

类型	描述
List[Message]	List of combined messages truncated if needed.

源代码位于： jianmu/message/retention.py

def apply(self, existing: List[Message], incoming: List[Message]) -> List[Message]:
    """Append incoming messages and truncate to the newest entries.

    Args:
        existing: List of existing messages.
        incoming: List of new incoming messages.

    Returns:
        List of combined messages truncated if needed.
    """
    merged = list(existing) + list(incoming)
    if self.max_messages is not None and self.max_messages > 0 and len(merged) > self.max_messages:
        return merged[-self.max_messages :]
    return merged

StateMessageStore

StateMessageStore(
    state_manager: "StateManager",
    max_messages: int | None = None,
    *,
    key: str = "messages",
    namespace: str | None = None,
    coercer: MessageCoercerProtocol | None = None,
    retention: RetentionPolicyProtocol | None = None,
    merge_fn: MergeFnProtocol | None = None,
)

Bases: MessageStoreProtocol

Message store backed by StateManager with pluggable coercion/retention.

This is the default in-process message store for Jianmu agents. It keeps chat history inside a state field, applies message coercion on writes, and optionally truncates or custom-merges history.

Typical usage::

store = StateMessageStore(sm, key="messages", max_messages=50)
store.append(human("hello"))
recent = store.get_messages(limit=10)

属性：

名称	类型	描述
_state_manager		Backing state manager storing the message list.
_key		State field key used for message history.
_namespace		Optional namespace applied to state operations.
_coercer		Message coercer used to normalize writes.
_merge_fn	MergeFnProtocol | None	Optional merge function applied during appends.
_max_messages	int | None	Optional tail-retention cap for stored history.
_subscribers	list[Callable[[MessageEvent], None]]	Registered message-event subscribers.
_sub_lock		Lock protecting subscriber registration and iteration.

Initialize a StateMessageStore.

参数：

名称	类型	描述	默认
state_manager	'StateManager'	The backing StateManager instance.	必需
max_messages	int | None	Maximum number of messages to retain (tail truncation). If both merge_fn and max_messages are provided, truncation is applied after merge_fn returns.	None
key	str	Field name within the schema (default: "messages").	'messages'
namespace	str | None	Optional runtime namespace prefix for key routing. Use this when each sub-agent or subtree needs isolated chat history.	None
coercer	MessageCoercerProtocol | None	Pluggable input normalization (default: StrictMessageCoercer).	None
retention	RetentionPolicyProtocol | None	High-level retention policy (mutually exclusive with merge_fn).	None
merge_fn	MergeFnProtocol | None	Low-level merge function. Takes (current, incoming) -> merged. Must be a sync function. Mutually exclusive with retention.	None

源代码位于： jianmu/message/store.py

def __init__(
    self,
    state_manager: "StateManager",
    max_messages: int | None = None,
    *,
    key: str = "messages",
    namespace: str | None = None,
    coercer: MessageCoercerProtocol | None = None,
    retention: RetentionPolicyProtocol | None = None,
    merge_fn: MergeFnProtocol | None = None,
):
    """Initialize a StateMessageStore.

    Args:
        state_manager: The backing StateManager instance.
        max_messages: Maximum number of messages to retain (tail truncation).
            If both merge_fn and max_messages are provided, truncation is applied
            after merge_fn returns.
        key: Field name within the schema (default: ``"messages"``).
        namespace: Optional runtime namespace prefix for key routing. Use
            this when each sub-agent or subtree needs isolated chat history.
        coercer: Pluggable input normalization (default: StrictMessageCoercer).
        retention: High-level retention policy (mutually exclusive with merge_fn).
        merge_fn: Low-level merge function. Takes (current, incoming) -> merged.
            Must be a sync function. Mutually exclusive with retention.
    """
    self._state_manager = state_manager
    self._key = key
    self._namespace = namespace
    self._coercer = coercer or StrictMessageCoercer()
    if retention is not None and max_messages is not None:
        raise ValueError("Do not pass both 'retention' and 'max_messages'. Configure retention in one place.")
    if retention is not None and merge_fn is not None:
        raise ValueError("Do not pass both 'retention' and 'merge_fn'. Configure merge strategy in one place.")
    if merge_fn is not None and inspect.iscoroutinefunction(merge_fn):
        raise TypeError("merge_fn must be a sync function. For async I/O, implement an async store backend.")

    if retention is not None:
        self._merge_fn: MergeFnProtocol | None = retention.apply
        self._max_messages: int | None = None
    else:
        self._merge_fn = merge_fn
        self._max_messages = max_messages

    self._subscribers: list[Callable[[MessageEvent], None]] = []
    self._sub_lock = threading.Lock()

append

append(message: Any) -> None

Append one message through the configured coercion pipeline.

参数：

名称	类型	描述	默认
message	Any	Message object or payload to be coerced and appended.	必需

源代码位于： jianmu/message/store.py

def append(self, message: Any) -> None:
    """Append one message through the configured coercion pipeline.

    Args:
        message: Message object or payload to be coerced and appended.
    """
    self.append_many([message])

append_many

append_many(messages: Sequence[Any]) -> None

Append multiple messages and emit per-message added events.

参数：

名称	类型	描述	默认
messages	Sequence[Any]	Sequence of message objects or payloads to coerce and append.	必需

源代码位于： jianmu/message/store.py

def append_many(self, messages: Sequence[Any]) -> None:
    """Append multiple messages and emit per-message added events.

    Args:
        messages: Sequence of message objects or payloads to coerce and append.
    """
    self._append_many(messages, signal=True)

replace

replace(
    messages: Sequence[Any], *, signal: bool = True
) -> None

Replace the full stored history with normalized messages.

参数：

名称	类型	描述	默认
messages	Sequence[Any]	Sequence of message objects or payloads to replace with.	必需
signal	bool	Whether to trigger change events on the state manager.	True

源代码位于： jianmu/message/store.py

def replace(self, messages: Sequence[Any], *, signal: bool = True) -> None:
    """Replace the full stored history with normalized messages.

    Args:
        messages: Sequence of message objects or payloads to replace with.
        signal: Whether to trigger change events on the state manager.
    """
    incoming = self._coerce_messages(list(messages))
    self._state_manager.transform_field(
        self._key,
        lambda _current: list(incoming),
        namespace=self._namespace,
        signal=signal,
    )

get_messages

get_messages(limit: int | None = None) -> List[Message]

Fetch coerced messages from the backing state store.

参数：

名称	类型	描述	默认
limit	int | None	Maximum number of messages to return from the tail of the store.	None

返回：

类型	描述
List[Message]	A list of coerced Message instances.

源代码位于： jianmu/message/store.py

def get_messages(self, limit: int | None = None) -> List[Message]:
    """Fetch coerced messages from the backing state store.

    Args:
        limit: Maximum number of messages to return from the tail of the store.

    Returns:
        A list of coerced Message instances.
    """
    raw = self._state_manager.get(self._key, namespace=self._namespace, default=[])
    messages = self._coerce_messages(list(raw)) if isinstance(raw, list) else []
    if limit and limit > 0 and len(messages) > limit:
        return messages[-limit:]
    return messages

read_input

read_input(
    raw: Any,
    *,
    fallback_role: str = "user",
    limit: int | None = None,
) -> List[Message]

Normalize one raw state payload into message objects.

参数：

名称	类型	描述	默认
raw	Any	Raw state payload read from the resolved input key.	必需
fallback_role	str	Role used when wrapping scalar payloads.	'user'
limit	int | None	Optional tail limit applied after normalization.	None

返回：

类型	描述
List[Message]	A normalized list of Message objects.

源代码位于： jianmu/message/store.py

def read_input(
    self,
    raw: Any,
    *,
    fallback_role: str = "user",
    limit: int | None = None,
) -> List[Message]:
    """Normalize one raw state payload into message objects.

    Args:
        raw: Raw state payload read from the resolved input key.
        fallback_role: Role used when wrapping scalar payloads.
        limit: Optional tail limit applied after normalization.

    Returns:
        A normalized list of ``Message`` objects.
    """
    if isinstance(raw, list):
        messages = self._coerce_messages(list(raw))
    else:
        messages = normalize_messages(raw, fallback_role=fallback_role)
    if limit and limit > 0 and len(messages) > limit:
        return messages[-limit:]
    return messages

subscribe

subscribe(
    callback: Callable[[MessageEvent], None],
) -> Callable[[MessageEvent], None]

Register a message event subscriber.

参数：

名称	类型	描述	默认
callback	Callable[[MessageEvent], None]	The callback function to subscribe.	必需

返回：

类型	描述
Callable[[MessageEvent], None]	The subscribed callback function.

源代码位于： jianmu/message/store.py

def subscribe(self, callback: Callable[[MessageEvent], None]) -> Callable[[MessageEvent], None]:
    """Register a message event subscriber.

    Args:
        callback: The callback function to subscribe.

    Returns:
        The subscribed callback function.
    """
    with self._sub_lock:
        self._subscribers.append(callback)
    return callback

unsubscribe

unsubscribe(
    callback: Callable[[MessageEvent], None],
) -> None

Remove a registered message event subscriber.

参数：

名称	类型	描述	默认
callback	Callable[[MessageEvent], None]	The callback function to unsubscribe.	必需

源代码位于： jianmu/message/store.py

def unsubscribe(self, callback: Callable[[MessageEvent], None]) -> None:
    """Remove a registered message event subscriber.

    Args:
        callback: The callback function to unsubscribe.
    """
    with self._sub_lock:
        try:
            self._subscribers.remove(callback)
        except ValueError:
            return

system

system(content: str, **kwargs: Any) -> Message

Create a system-role message.

参数：

名称	类型	描述	默认
content	str	Message body text.	必需
**kwargs	Any	Additional Message field overrides.	{}

返回：

类型	描述
Message	A Message instance with role='system'.

These helper factories are primarily for readability in examples and tests.

源代码位于： jianmu/message/base.py

def system(content: str, **kwargs: Any) -> Message:
    """Create a system-role message.

    Args:
        content: Message body text.
        **kwargs: Additional ``Message`` field overrides.

    Returns:
        A ``Message`` instance with ``role='system'``.

    These helper factories are primarily for readability in examples and tests.
    """
    return Message(role="system", content=content, **kwargs)

human

human(content: str, **kwargs: Any) -> Message

Create a user-role message.

参数：

名称	类型	描述	默认
content	str	Message body text.	必需
**kwargs	Any	Additional Message field overrides.	{}

返回：

类型	描述
Message	A Message instance with role='user'.

源代码位于： jianmu/message/base.py

def human(content: str, **kwargs: Any) -> Message:
    """Create a user-role message.

    Args:
        content: Message body text.
        **kwargs: Additional ``Message`` field overrides.

    Returns:
        A ``Message`` instance with ``role='user'``.
    """
    return Message(role="user", content=content, **kwargs)

ai

ai(content: str, **kwargs: Any) -> Message

Create an assistant-role message.

参数：

名称	类型	描述	默认
content	str	Message body text.	必需
**kwargs	Any	Additional Message field overrides.	{}

返回：

类型	描述
Message	A Message instance with role='assistant'.

源代码位于： jianmu/message/base.py

def ai(content: str, **kwargs: Any) -> Message:
    """Create an assistant-role message.

    Args:
        content: Message body text.
        **kwargs: Additional ``Message`` field overrides.

    Returns:
        A ``Message`` instance with ``role='assistant'``.
    """
    return Message(role="assistant", content=content, **kwargs)

tool

tool(
    content: str, name: Optional[str] = None, **kwargs: Any
) -> Message

Create a tool-role message.

参数：

名称	类型	描述	默认
content	str	Tool observation or output text.	必需
name	Optional[str]	Optional tool name associated with this message.	None
**kwargs	Any	Additional Message field overrides.	{}

返回：

类型	描述
Message	A Message instance with role='tool'.

源代码位于： jianmu/message/base.py

def tool(content: str, name: Optional[str] = None, **kwargs: Any) -> Message:
    """Create a tool-role message.

    Args:
        content: Tool observation or output text.
        name: Optional tool name associated with this message.
        **kwargs: Additional ``Message`` field overrides.

    Returns:
        A ``Message`` instance with ``role='tool'``.
    """
    return Message(role="tool", content=content, name=name, **kwargs)

normalize_message

normalize_message(
    payload: Any,
    *,
    fallback_role: str = "user",
    coercer: MessageCoercerProtocol | None = None,
) -> Message | None

Normalize one arbitrary payload into a Message.

参数：

名称	类型	描述	默认
payload	Any	Message-like payload to normalize.	必需
fallback_role	str	Role used when wrapping non-message payloads.	'user'
coercer	MessageCoercerProtocol | None	Optional explicit coercer override. When provided, its result takes precedence over the built-in best-effort normalization.	None

返回：

类型	描述
Message | None	A normalized Message instance or None when the payload is empty.

源代码位于： jianmu/message/coercer.py

def normalize_message(
    payload: Any,
    *,
    fallback_role: str = "user",
    coercer: MessageCoercerProtocol | None = None,
) -> Message | None:
    """Normalize one arbitrary payload into a ``Message``.

    Args:
        payload: Message-like payload to normalize.
        fallback_role: Role used when wrapping non-message payloads.
        coercer: Optional explicit coercer override. When provided, its result
            takes precedence over the built-in best-effort normalization.

    Returns:
        A normalized ``Message`` instance or ``None`` when the payload is empty.
    """
    if coercer is not None:
        return coercer.coerce(payload)
    if payload is None:
        return None
    if isinstance(payload, Message):
        return payload
    if isinstance(payload, dict):
        return Message.model_validate(payload)

    text = Message.to_text(payload).strip()
    if not text:
        return None
    return Message(role=fallback_role, content=text)

normalize_messages

normalize_messages(
    payload: Sequence[Any] | Any | None,
    *,
    fallback_role: str = "user",
    coercer: MessageCoercerProtocol | None = None,
) -> list[Message]

Normalize one message payload or a message sequence.

参数：

名称	类型	描述	默认
payload	Sequence[Any] | Any | None	Single message-like value, sequence of values, or None.	必需
fallback_role	str	Role used when wrapping non-message payloads.	'user'
coercer	MessageCoercerProtocol | None	Optional explicit coercer override applied item-by-item.	None

返回：

类型	描述
list[Message]	A list of normalized Message objects.

源代码位于： jianmu/message/coercer.py

def normalize_messages(
    payload: Sequence[Any] | Any | None,
    *,
    fallback_role: str = "user",
    coercer: MessageCoercerProtocol | None = None,
) -> list[Message]:
    """Normalize one message payload or a message sequence.

    Args:
        payload: Single message-like value, sequence of values, or ``None``.
        fallback_role: Role used when wrapping non-message payloads.
        coercer: Optional explicit coercer override applied item-by-item.

    Returns:
        A list of normalized ``Message`` objects.
    """
    if payload is None:
        return []

    items = list(payload) if isinstance(payload, Sequence) and not isinstance(payload, (str, bytes, dict, Message)) else [payload]
    normalized: list[Message] = []
    for item in items:
        coerced = normalize_message(
            item,
            fallback_role=fallback_role,
            coercer=coercer,
        )
        if coerced is not None:
            normalized.append(coerced)
    return normalized