feat(langchain|langgraph): add AGUI protocol support with async generator handlers and lifecycle hooks

This commit introduces comprehensive AGUI protocol support with lifecycle hooks for both LangChain and LangGraph integrations. The changes include:

1. Added convert function exports to both LangChain and LangGraph integration modules with usage examples
2. Refactored model adapter to use default_headers instead of async_client for ChatOpenAI
3. Enhanced server invoker to handle both coroutine and async generator functions
4. Updated protocol handlers to support async response formatting and safe iterator handling
5. Refined AgentRequest model to be protocol-agnostic with core fields
6. Added CORS middleware configuration to AgentRunServer
7. Updated quick start example to use async streaming with event conversion

The changes enable proper streaming support, tool call event handling, and improved async performance for both LangChain and LangGraph integrations.

feat(langchain|langgraph): 添加 AGUI 协议支持和异步生成器处理器以及生命周期钩子

此提交为 LangChain 和 LangGraph 集成引入了全面的 AGUI 协议支持和生命周期钩子。变更包括：

1. 为 LangChain 和 LangGraph 集成模块添加了 convert 函数导出以及使用示例
2. 重构模型适配器，使用 default_headers 替代 async_client 用于 ChatOpenAI
3. 增强服务器调用器以处理协程和异步生成器函数
4. 更新协议处理器以支持异步响应格式化和安全的迭代器处理
5. 精简 AgentRequest 模型以实现协议无关性并使用核心字段
6. 为 AgentRunServer 添加 CORS 中间件配置
7. 更新快速开始示例以使用带事件转换的异步流

这些变更启用了适当的流式传输支持、工具调用事件处理以及改进的异步性能。

Change-Id: I941d1d797b930243282555b5a6db0e6d420f3691
Signed-off-by: OhYee <oyohyee@oyohyee.com>

Author: OhYee

agentrun/integration/langchain/__init__.py

@@ -1,8 +1,24 @@
-"""LangChain 集成模块,提供 AgentRun 模型与沙箱的 LangChain 适配。 / LangChain 集成 Module"""
+"""LangChain 集成模块
+
+Example:
+    >>> from langchain.agents import create_agent
+    >>> from agentrun.integration.langchain import convert, model, toolset
+    >>>
+    >>> agent = create_agent(model=model("my-model"), tools=toolset("my-tools"))
+    >>>
+    >>> async def invoke_agent(request: AgentRequest):
+    ...     input_data = {"messages": [...]}
+    ...     async for event in agent.astream_events(input_data, version="v2"):
+    ...         for item in convert(event, request.hooks):
+    ...             yield item
+"""
+
+from agentrun.integration.langgraph.agent_converter import convert
 
 from .builtin import model, sandbox_toolset, toolset
 
 __all__ = [
+    "convert",
     "model",
     "toolset",
     "sandbox_toolset",

agentrun/integration/langchain/model_adapter.py

@@ -23,7 +23,6 @@ def __init__(self):
 
     def wrap_model(self, common_model: Any) -> Any:
         """包装 CommonModel 为 LangChain BaseChatModel / LangChain Model Adapter"""
-        from httpx import AsyncClient
         from langchain_openai import ChatOpenAI
 
         info = common_model.get_model_info()  # 确保模型可用
@@ -32,6 +31,6 @@ def wrap_model(self, common_model: Any) -> Any:
             api_key=info.api_key,
             model=info.model,
             base_url=info.base_url,
-            async_client=AsyncClient(headers=info.headers),
+            default_headers=info.headers,
             stream_usage=True,
         )

agentrun/integration/langgraph/__init__.py

@@ -1,12 +1,23 @@
-"""LangGraph 集成模块。 / LangGraph 集成 Module
+"""LangGraph 集成模块
 
-提供 AgentRun 模型与沙箱工具的 LangGraph 适配入口。 / 提供 AgentRun 模型with沙箱工具的 LangGraph 适配入口。
-LangGraph 与 LangChain 兼容，因此直接复用 LangChain 的转换逻辑。 / LangGraph with LangChain 兼容，因此直接复用 LangChain 的转换逻辑。
+Example:
+    >>> from langgraph.prebuilt import create_react_agent
+    >>> from agentrun.integration.langgraph import convert
+    >>>
+    >>> agent = create_react_agent(llm, tools)
+    >>>
+    >>> async def invoke_agent(request: AgentRequest):
+    ...     input_data = {"messages": [...]}
+    ...     async for event in agent.astream_events(input_data, version="v2"):
+    ...         for item in convert(event, request.hooks):
+    ...             yield item
 """
 
+from .agent_converter import convert
 from .builtin import model, sandbox_toolset, toolset
 
 __all__ = [
+    "convert",
     "model",
     "toolset",
     "sandbox_toolset",

agentrun/integration/langgraph/agent_converter.py

@@ -0,0 +1,177 @@
+"""LangGraph/LangChain Agent 事件转换器
+
+将 LangGraph/LangChain astream_events 的单个事件转换为 AgentRun 事件。
+
+支持两种事件格式：
+1. on_chat_model_stream - LangGraph create_react_agent 的流式输出
+2. on_chain_stream - LangChain create_agent 的输出
+
+Example:
+    >>> async def invoke_agent(request: AgentRequest):
+    ...     async for event in agent.astream_events(input_data, version="v2"):
+    ...         for item in convert(event, request.hooks):
+    ...             yield item
+"""
+
+import json
+from typing import Any, Dict, Generator, List, Optional, Union
+
+from agentrun.server.model import AgentEvent, AgentLifecycleHooks
+
+
+def convert(
+    event: Dict[str, Any],
+    hooks: Optional[AgentLifecycleHooks] = None,
+) -> Generator[Union[AgentEvent, str], None, None]:
+    """转换单个 astream_events 事件
+
+    Args:
+        event: LangGraph/LangChain astream_events 的单个事件
+        hooks: AgentLifecycleHooks，用于创建工具调用事件
+
+    Yields:
+        str (文本内容) 或 AgentEvent (工具调用事件)
+    """
+    event_type = event.get("event", "")
+    data = event.get("data", {})
+
+    # 1. LangGraph 格式: on_chat_model_stream
+    if event_type == "on_chat_model_stream":
+        chunk = data.get("chunk")
+        if chunk:
+            content = _get_content(chunk)
+            if content:
+                yield content
+
+            # 流式工具调用参数
+            if hooks:
+                for tc in _get_tool_chunks(chunk):
+                    tc_id = tc.get("id") or str(tc.get("index", ""))
+                    if tc.get("name") and tc_id:
+                        yield hooks.on_tool_call_start(
+                            id=tc_id, name=tc["name"]
+                        )
+                    if tc.get("args") and tc_id:
+                        yield hooks.on_tool_call_args_delta(
+                            id=tc_id, delta=tc["args"]
+                        )
+
+    # 2. LangChain 格式: on_chain_stream (来自 create_agent)
+    #    只处理 name="model" 的事件，避免重复（LangGraph 会发送 name="model" 和 name="LangGraph" 两个相同内容的事件）
+    elif event_type == "on_chain_stream" and event.get("name") == "model":
+        chunk_data = data.get("chunk", {})
+        if isinstance(chunk_data, dict):
+            # chunk 格式: {"messages": [AIMessage(...)]}
+            messages = chunk_data.get("messages", [])
+
+            for msg in messages:
+                # 提取文本内容
+                content = _get_content(msg)
+                if content:
+                    yield content
+
+                # 提取工具调用
+                if hooks:
+                    tool_calls = _get_tool_calls(msg)
+                    for tc in tool_calls:
+                        tc_id = tc.get("id", "")
+                        tc_name = tc.get("name", "")
+                        tc_args = tc.get("args", {})
+                        if tc_id and tc_name:
+                            yield hooks.on_tool_call_start(
+                                id=tc_id, name=tc_name
+                            )
+                            if tc_args:
+                                yield hooks.on_tool_call_args(
+                                    id=tc_id, args=_to_json(tc_args)
+                                )
+
+    # 3. 工具开始 (LangGraph)
+    elif event_type == "on_tool_start" and hooks:
+        run_id = event.get("run_id", "")
+        tool_name = event.get("name", "")
+        tool_input = data.get("input", {})
+
+        if run_id:
+            yield hooks.on_tool_call_start(id=run_id, name=tool_name)
+            if tool_input:
+                yield hooks.on_tool_call_args(
+                    id=run_id, args=_to_json(tool_input)
+                )
+
+    # 4. 工具结束 (LangGraph)
+    elif event_type == "on_tool_end" and hooks:
+        run_id = event.get("run_id", "")
+        output = data.get("output", "")
+
+        if run_id:
+            yield hooks.on_tool_call_result(
+                id=run_id, result=str(output) if output else ""
+            )
+            yield hooks.on_tool_call_end(id=run_id)
+
+
+def _get_content(obj: Any) -> Optional[str]:
+    """提取文本内容"""
+    if obj is None:
+        return None
+
+    # 字符串
+    if isinstance(obj, str):
+        return obj if obj else None
+
+    # 有 content 属性的对象 (AIMessage, AIMessageChunk, etc.)
+    if hasattr(obj, "content"):
+        c = obj.content
+        if isinstance(c, str) and c:
+            return c
+        if isinstance(c, list):
+            parts = []
+            for item in c:
+                if isinstance(item, str):
+                    parts.append(item)
+                elif isinstance(item, dict):
+                    parts.append(item.get("text", ""))
+            return "".join(parts) or None
+
+    return None
+
+
+def _get_tool_chunks(chunk: Any) -> List[Dict[str, Any]]:
+    """提取工具调用增量 (AIMessageChunk.tool_call_chunks)"""
+    result: List[Dict[str, Any]] = []
+    if hasattr(chunk, "tool_call_chunks") and chunk.tool_call_chunks:
+        for tc in chunk.tool_call_chunks:
+            if isinstance(tc, dict):
+                result.append(tc)
+            else:
+                result.append({
+                    "id": getattr(tc, "id", None),
+                    "name": getattr(tc, "name", None),
+                    "args": getattr(tc, "args", None),
+                    "index": getattr(tc, "index", None),
+                })
+    return result
+
+
+def _get_tool_calls(msg: Any) -> List[Dict[str, Any]]:
+    """提取完整工具调用 (AIMessage.tool_calls)"""
+    result: List[Dict[str, Any]] = []
+    if hasattr(msg, "tool_calls") and msg.tool_calls:
+        for tc in msg.tool_calls:
+            if isinstance(tc, dict):
+                result.append(tc)
+            else:
+                result.append({
+                    "id": getattr(tc, "id", None),
+                    "name": getattr(tc, "name", None),
+                    "args": getattr(tc, "args", None),
+                })
+    return result
+
+
+def _to_json(obj: Any) -> str:
+    """转 JSON 字符串"""
+    if isinstance(obj, str):
+        return obj
+    return json.dumps(obj, ensure_ascii=False)

agentrun/server/__init__.py

@@ -1,46 +1,54 @@
 """AgentRun Server 模块 / AgentRun Server Module
 
-提供 HTTP Server 集成能力,支持符合 AgentRun 规范的 Agent 调用接口。
+提供 HTTP Server 集成能力，支持符合 AgentRun 规范的 Agent 调用接口。
 支持 OpenAI Chat Completions 和 AG-UI 两种协议。
 
-Example (基本使用 - 同步):
+Example (基本使用 - 返回字符串):
 >>> from agentrun.server import AgentRunServer, AgentRequest
 >>>
 >>> def invoke_agent(request: AgentRequest):
-...     # 实现你的 Agent 逻辑
 ...     return "Hello, world!"
 >>>
 >>> server = AgentRunServer(invoke_agent=invoke_agent)
->>> server.start(host="0.0.0.0", port=8080)
+>>> server.start(port=9000)
 
-Example (使用生命周期钩子 - 同步，推荐):
+Example (流式输出):
+>>> def invoke_agent(request: AgentRequest):
+...     for word in ["Hello", ", ", "world", "!"]:
+...         yield word
+>>>
+>>> AgentRunServer(invoke_agent=invoke_agent).start()
+
+Example (使用生命周期钩子):
 >>> def invoke_agent(request: AgentRequest):
 ...     hooks = request.hooks
 ...
-...     # 发送步骤开始事件 (使用 emit_* 同步方法)
-...     yield hooks.emit_step_start("processing")
+...     # 发送步骤开始事件
+...     yield hooks.on_step_start("processing")
 ...
-...     # 处理逻辑...
+...     # 流式输出内容
 ...     yield "Hello, "
 ...     yield "world!"
 ...
 ...     # 发送步骤结束事件
-...     yield hooks.emit_step_finish("processing")
+...     yield hooks.on_step_finish("processing")
 
-Example (使用生命周期钩子 - 异步):
->>> async def invoke_agent(request: AgentRequest):
+Example (工具调用事件):
+>>> def invoke_agent(request: AgentRequest):
 ...     hooks = request.hooks
 ...
-...     # 发送步骤开始事件 (使用 on_* 异步方法)
-...     async for event in hooks.on_step_start("processing"):
-...         yield event
+...     # 工具调用开始
+...     yield hooks.on_tool_call_start(id="call_1", name="get_time")
+...     yield hooks.on_tool_call_args(id="call_1", args={"timezone": "UTC"})
 ...
-...     # 处理逻辑...
-...     yield "Hello, world!"
+...     # 执行工具
+...     result = "2024-01-01 12:00:00"
 ...
-...     # 发送步骤结束事件
-...     async for event in hooks.on_step_finish("processing"):
-...         yield event
+...     # 工具调用结果
+...     yield hooks.on_tool_call_result(id="call_1", result=result)
+...     yield hooks.on_tool_call_end(id="call_1")
+...
+...     yield f"当前时间: {result}"
 
 Example (访问原始请求):
 >>> def invoke_agent(request: AgentRequest):

agentrun/server/agui_protocol.py

@@ -14,6 +14,7 @@
 """
 
 from enum import Enum
+import inspect
 import json
 import time
 from typing import (
@@ -536,6 +537,9 @@ async def run_agent(request: Request):
                 event_stream = self.format_response(
                     agent_result, agent_request, context
                 )
+                # 支持 format_response 返回 coroutine 或者 async iterator
+                if inspect.isawaitable(event_stream):
+                    event_stream = await event_stream
 
                 # 4. 返回 SSE 流
                 return StreamingResponse(
@@ -779,13 +783,20 @@ async def _iterate_content(
             loop = asyncio.get_event_loop()
             iterator = iter(content)  # type: ignore
 
-            while True:
+            # 使用哨兵值来检测迭代结束，避免 StopIteration 传播到 Future
+            _STOP = object()
+
+            def _safe_next():
                 try:
-                    # 在线程池中执行 next()，避免 time.sleep 阻塞事件循环
-                    chunk = await loop.run_in_executor(None, next, iterator)
-                    yield chunk
+                    return next(iterator)
                 except StopIteration:
+                    return _STOP
+
+            while True:
+                chunk = await loop.run_in_executor(None, _safe_next)
+                if chunk is _STOP:
                     break
+                yield chunk
 
 
 # ============================================================================