跳转至

progress

feishu.agent.progress

ProgressSnapshot dataclass

供产品侧生成可见进度文案的短暂快照。

reasoning 可包含模型流式输出的原始近期 thinking/reasoning 片段。SDK 只在内存中传给 progress_summarizer,不落会话历史、审批审计或日志;产品侧应把它摘要为可展示的一句话。

源代码位于: feishu/agent/progress.py
Python
@dataclass(frozen=True)
class ProgressSnapshot:
    r"""
    供产品侧生成可见进度文案的短暂快照。

    `reasoning` 可包含模型流式输出的原始近期 thinking/reasoning 片段。SDK 只在内存中传给
    `progress_summarizer`,不落会话历史、审批审计或日志;产品侧应把它摘要为可展示的一句话。
    """

    phase: str
    reasoning: str = ""
    text: str = ""
    tool_name: str | None = None
    tool_description: str | None = None
    elapsed_seconds: float = 0.0

build_progress_summarizer

Python
build_progress_summarizer(backend: Any, *, timeout_seconds: float = 1.2, max_chars: int = 60, instruction: str = PROGRESS_SUMMARY_INSTRUCTION) -> Any

基于通用文本摘要器构建一次性的进度文案摘要器。

源代码位于: feishu/agent/progress.py
Python
def build_progress_summarizer(
    backend: Any,
    *,
    timeout_seconds: float = 1.2,
    max_chars: int = 60,
    instruction: str = PROGRESS_SUMMARY_INSTRUCTION,
) -> Any:
    r"""基于通用文本摘要器构建一次性的进度文案摘要器。"""
    from .summarization import TextSummaryRequest, build_fast_text_summarizer

    text_summarizer = build_fast_text_summarizer(
        backend,
        timeout_seconds=timeout_seconds,
        default_max_chars=max_chars,
    )

    async def summarize(snapshot: ProgressSnapshot) -> str | None:
        prompt = _prompt(snapshot, max_chars=max_chars)
        result = await text_summarizer(
            TextSummaryRequest(
                kind="progress",
                instruction=instruction,
                text=prompt,
                max_chars=max_chars,
            )
        )
        status = _clean_status(result or "", max_chars=max_chars)
        if status is None:
            logger.info("progress summarizer returned empty text; using fallback")
            return _fallback_status(snapshot, max_chars=max_chars)
        return status

    return summarize