Context 模块

核心文件

• src/by_framework/worker/context.py - AgentContext
• src/by_framework/worker/byai_context.py - ByaiAgentContext

AgentContext

运行时上下文，用于任务执行过程中与环境交互。

属性

属性	类型	描述
session_id	str	会话ID
trace_id	str	追踪ID
message_id	str	消息ID
parent_message_id	str	父消息ID
current_agent_id	str	当前Agent ID

核心方法

emit_chunk

发送流式文本片段：

async def emit_chunk(
    self,
    event: Union[StreamChunkEvent, str],
    event_type: Optional[str] = None
) -> None:

emit_state

发送状态更新：

async def emit_state(
    self,
    event: Union[StateChangeEvent, str],
    event_type: Optional[str] = None
) -> None:

emit_artifact

发送产物/附件：

async def emit_artifact(
    self,
    event: Union[ArtifactEvent, str],
    event_type: Optional[str] = None
) -> None:

call_agent

调用其他Agent：

async def call_agent(
    self,
    target_agent_type: str,
    content: object,
    **kwargs
) -> dict:

ask_user

向用户请求输入（挂起当前任务）：

async def ask_user(
    self,
    event: Union[AskUserEvent, str]
) -> dict:

dispatch_group

分发任务组：

async def dispatch_group(
    self,
    tasks: list[dict],
    **kwargs
) -> dict:

ByaiAgentContext

继承自 AgentContext，配合 BaiYing message 对象使用，框架自动做内容编解码。