leak problem

async for event in runner.run_async(
            user_id=user_id,
            session_id=session_id,
            new_message=new_message,
            state_delta=state_delta,
            run_config=run_config, # Stream=True模式
        ):
            if some_condition:
                   return None`

Problem: When the runner's events are not fully iterated (ending prematurely), it can cause a leak in the LLM request connection. Below is the test code based on LiteLLM. This code is consistent with the streaming output invocation method currently used in the ADK↓↓↓

async def test_acompletion_resource_cleanup(end):
    # Make an async completion call
    cnt = 1
    async for el in await litellm.acompletion(
        model="deepseek/deepseek-chat",
        api_key="...",
        base_url="https://api.deepseek.com/v1",
        messages=[
            {
                "role": "user",
                "content": "一次不差的复述下下方短文内容：在这个信息爆炸的时代，静下心来读一本书，成了一种奢侈。手机、社交媒体、短视频接连不断地抢占我们的注意力，使我们越来越难以集中精力去做一件事情。很多人开始感到焦虑，仿佛每天都很忙，却不知忙了些什么。其实，我们真正需要的，是一种“慢下来”的勇气。在快节奏的生活中，抽出时间去散步、阅读、与亲人交流，这些看似平凡的小事，恰恰是我们获得内心安宁的关键。当我们暂时放下外界的喧嚣，才更容易听见内心的声音。慢，不是懒惰，而是一种自我掌控的表现。它让我们重新掌握生活的节奏，不再被外界牵着走。因此，学会在忙碌中寻找片刻的安静，是每个人都该修炼的生活智慧。",
            }
        ],
        stream=True,
    ):
        print(f"{end} {el}")
        if end:
            cnt += 1
            if cnt > 3:
                break

    # await litellm.close_litellm_async_clients()
    # lsof -i @api.deepseek.com | grep -c "https"


async def main():
    # Run the test
    """
    await test_acompletion_resource_cleanup(True)
    await test_acompletion_resource_cleanup(False)
    print("✅ All tests passed!")

    """

    # Run the tests concurrently
    await asyncio.gather(
        test_acompletion_resource_cleanup(True),
        test_acompletion_resource_cleanup(False)
    )
    print("✅ All tests passed!")

if __name__ == "__main__":
    asyncio.run(main())

Here's a personal perspective:

When adk calls lite_llm and obtains a CustomStreamWrapper, if it returns prematurely, the CustomStreamWrapper cannot correctly execute the final release logic because it lacks aclose and aexit methods—it relies on anext for this purpose.

如果你遇到AttributeError: 'Choices'对象没有'logprobs'属性的错误，可以修改代码为逐个使用await，避免使用gather。

@hangfei

@hangfei
Hello, the context for my question is that I have a task with a lengthy automated process that outputs critical information to users. Users can STOP at any time or chat with the AI after stopping and then resume the AI's work. I want to achieve this functionality with just one STOP capability, similar to most LLM software, especially in stream mode.

Is there any plan for the ADK to support a "stop" feature? Below is my temporary implementation, but I'm unsure if there are any HTTP leakage issues, even after reviewing the code and consulting DeepWiki (ADK and LiteLLM).

adk version=1.5.0

class MyRunner(ABC, Runner):

    _RUNNING = "running"
    _STOPPING = "stopping"
    _READY = "ready"

    def __init__(
        self,
        *,
        app_name: str,
        agent: BaseAgent,
        artifact_service: Optional[BaseArtifactService] = None,
        session_service: BaseSessionService,
        memory_service: Optional[BaseMemoryService] = None,
        credential_service: Optional[BaseCredentialService] = None,
    ):
        super().__init__(
            app_name=app_name,
            agent=agent,
            artifact_service=artifact_service,
            session_service=session_service,
            memory_service=memory_service,
            credential_service=credential_service,
        )

    @abstractmethod
    async def get_session_stop_status(self, user_id: str, session_id: str) -> str:
        """
        方法实现要求：
            - stream模式下，该接口必须响应极快否则chunk的内容推送会延迟过高影响体验
            - stream模式下，该接口必须能抗住高qps
        """
        raise NotImplementedError("方法实现必须响应极快避免chunk内容延迟过高")

    @abstractmethod
    async def set_session_stop_status(self, user_id: str, session_id: str, status: str) -> None:
        raise NotImplementedError()

    async def run_async(
        self,
        *,
        user_id: str,
        session_id: str,
        new_message: types.Content,
        run_config: RunConfig = RunConfig(),
    ) -> AsyncGenerator[Event, None]:

        stop_final_text = ""
        await self.set_session_stop_status(user_id, session_id, MyRunner._RUNNING)

        async for event in super().run_async(
            user_id=user_id,
            session_id=session_id,
            new_message=new_message,
            run_config=run_config,
        ):

            """
            stop logic
            """
            if not event.is_final_response():

                if event.partial:
                    for part in event.content.parts:  # type: ignore[union-attr]
                        stop_final_text += part.text  # type: ignore[operator]
                else:
                    # 理论上不会触发，这里是为了安全：避免底层代码调整导致的逻辑错误
                    stop_final_text = ""

                is_not_tool_call_event = (
                    len(event.get_function_calls()) == 0
                )  # 考虑到上下文的完整性，如果消息为tool_call事件，则不参与stop

                if (
                    is_not_tool_call_event
                    and await self.get_session_stop_status(user_id, session_id)
                    == MyRunner._STOPPING
                ):

                    stop_final_event = None
                    if event.partial:  # 一定是个text chunk
                        stop_final_event = event.__deepcopy__()
                        stop_final_event.content.parts = [  # type: ignore[union-attr]
                            Part.from_text(text=stop_final_text)
                        ]
                        stop_final_event.partial = False

                        session = await self.session_service.get_session(
                            app_name=self.app_name,
                            user_id=user_id,
                            session_id=session_id,
                        )

                        if session is None:
                            raise BizException(f"Session not found for app_name={self.app_name}, user_id={user_id}, session_id={session_id}")

                        await self.session_service.append_event(
                            session=session, event=stop_final_event
                        )

                    await self.set_session_stop_status(
                        user_id, session_id, MyRunner._READY
                    )

                    # 注意：最后1条消息需要是final response 这样前台业务才会自适应终止（或许未来会补发1条特殊的消息来告诉模型 用户STOP过 ）
                    if stop_final_event:  # 一定是个text chunk
                        yield event
                        yield stop_final_event
                    else:  # 一定是个function response
                        event.actions.skip_summarization = True
                        yield event

                    lp_logger.debug_log.info(
                        f"{self.app_name} {user_id} {session_id} 已停止 last_event\n {event} \n agg_stop_event\n {stop_final_event}"
                    )
                    return
            else:
                stop_final_text = ""
                await self.set_session_stop_status(user_id, session_id, MyRunner._READY)

            yield event

    async def stop(self, user_id: str, session_id: str) -> None:

        stop_status = await self.get_session_stop_status(user_id, session_id)
        if (
            not stop_status
            or stop_status == MyRunner._STOPPING
            or stop_status == MyRunner._READY
        ):
            lp_logger.debug_log.info(
                f"stop - {user_id}_{session_id} by status {stop_status}"
            )
            return None

        await self.set_session_stop_status(user_id, session_id, MyRunner._STOPPING)
        lp_logger.debug_log.info(f"stop - {user_id}_{session_id}")
        return None

Navigation Menu

Repository navigation

leak problem #2068

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

leak problem #2068

Description

Activity

growmuye commented on Jul 24, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions