Microsoft Agent Framework 领域模型

@runtime_checkable
class AgentProtocol(Protocol):
    """Agent 协议定义 - 结构化子类型支持"""
    
    # 标识属性
    id: str
    name: str | None
    description: str | None
    
    # 核心方法
    async def run(
        self,
        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
        *,
        thread: AgentThread | None = None,
        **kwargs: Any,
    ) -> AgentResponse: ...
    
    def run_stream(
        self,
        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
        *,
        thread: AgentThread | None = None,
        **kwargs: Any,
    ) -> AsyncIterable[AgentResponseUpdate]: ...
    
    def get_new_thread(self, **kwargs: Any) -> AgentThread: ...

@runtime_checkable
class ChatClientProtocol(Protocol[TOptions_contra]):
    """Chat 客户端协议 - LLM 服务抽象"""
    
    additional_properties: dict[str, Any]
    
    async def get_response(
        self,
        messages: MutableSequence[ChatMessage],
        *,
        options: TOptions_contra | None = None,
        **kwargs: Any,
    ) -> ChatResponse: ...
    
    def get_streaming_response(
        self,
        messages: MutableSequence[ChatMessage],
        *,
        options: TOptions_contra | None = None,
        **kwargs: Any,
    ) -> AsyncIterable[ChatResponseUpdate]: ...

@runtime_checkable
class ToolProtocol(Protocol):
    """工具协议定义"""
    
    name: str
    description: str
    additional_properties: dict[str, Any] | None
    
    def __str__(self) -> str: ...

class ContextProvider(Protocol):
    """上下文提供者协议 - 动态注入"""
    
    async def invoking(
        self,
        messages: ChatMessage | Sequence[ChatMessage],
        **kwargs: Any,
    ) -> Context | None: ...
    
    async def invoked(
        self,
        input_messages: ChatMessage | Sequence[ChatMessage],
        response_messages: ChatMessage | Sequence[ChatMessage],
        **kwargs: Any,
    ) -> None: ...
    
    def thread_created(self, thread_id: str) -> None: ...

class AgentMiddleware(ABC):
    """Agent 中间件抽象基类"""
    
    @abstractmethod
    async def process(
        self,
        context: AgentRunContext,
        next: Callable[[AgentRunContext], Awaitable[None]],
    ) -> None:
        """
        Process the agent run context.
        
        Args:
            context: The agent run context containing agent, messages, thread, etc.
            next: The next middleware in the chain or the final handler.
        
        Note:
            - Call `await next(context)` to continue the chain
            - Set `context.terminate = True` to stop further processing
            - Set `context.result` to override the result
        """
        ...

# type 字段必须是 ContentType 字面量之一
type: ContentType  # Literal["text", "data", "uri", ...]

# function_call 和 function_result 必须包含 call_id
Content.from_function_call(call_id="c1", name="test")  # ✓
Content.from_function_call(name="test")  # ✗ - call_id 必需

# 文本内容加法要求相同 call_id
Content.from_function_call(call_id="c1", name="n", arguments='{"a":') + \
Content.from_function_call(call_id="c1", name="n", arguments='1}')  # ✓

# role 必须是 Role 枚举值
ChatMessage(role=Role.USER, text="Hello")  # ✓
ChatMessage(role="user", text="Hello")  # 自动转换为 Role.USER

# text 和 contents 互斥（内部转换）
ChatMessage(role=Role.USER, text="Hello")  # 内部创建 Content.from_text
ChatMessage(role=Role.USER, contents=[Content.from_text("Hello")])  # 等效

# input_model 必须是 Pydantic BaseModel 子类
FunctionTool(name="t", func=f, input_model=MyInputModel)  # ✓
FunctionTool(name="t", func=f, input_model=dict)  # ✗ - 必须是 BaseModel

# approval_mode 必须是指定字面量
approval_mode: Literal["always_require", "never_require"] | None

# SCHEMA_VERSION 用于状态迁移
schema_version: str  # 如 "1.1.0"

# conversation_history 按时间顺序排列
conversation_history: list[DurableAgentStateEntry]  # Request -> Response -> Request -> ...

# correlation_id 唯一标识请求/响应对
try_get_agent_response(correlation_id: str) -> AgentResponse | None