feat(langgraph|langchain): introduce AgentRunConverter for consistent tool_call_id handling

adds new AgentRunConverter class to maintain tool_call_id consistency in streaming tool calls,
where the first chunk provides the id and subsequent chunks only have index. also updates
integration modules to use the new converter class and maintains backward compatibility.

The new AgentRunConverter class manages the mapping between tool call indices and IDs to ensure
consistent tool_call_id across streaming chunks, which is crucial for proper tool call
correlation in the AG-UI protocol.

// 中文翻译：

添加了新的 AgentRunConverter 类来维护流式工具调用中 tool_call_id 的一致性，
其中第一个数据块提供 ID，后续的数据块只有索引。同时更新了集成模块以使用新
的转换器类并保持向后兼容性。

新的 AgentRunConverter 类管理工具调用索引和 ID 之间的映射，确保跨流式数据块的
tool_call_id 一致性，这对于 AG-UI 协议中的正确工具调用关联至关重要。

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

Author: OhYee

agentrun/integration/langchain/__init__.py

@@ -1,13 +1,13 @@
 """LangChain 集成模块
 
-使用 to_agui_events 将 LangChain 事件转换为 AG-UI 协议事件：
+使用 AgentRunConverter 将 LangChain 事件转换为 AG-UI 协议事件：
 
-    >>> from agentrun.integration.langchain import to_agui_events
+    >>> from agentrun.integration.langchain import AgentRunConverter
     >>>
     >>> async def invoke_agent(request: AgentRequest):
-    ...     input_data = {"messages": [...]}
+    ...     converter = AgentRunConverter()
     ...     async for event in agent.astream_events(input_data, version="v2"):
-    ...         for item in to_agui_events(event):
+    ...         for item in converter.convert(event):
     ...             yield item
 
 支持多种调用方式：
@@ -17,15 +17,21 @@
 """
 
 from agentrun.integration.langgraph.agent_converter import (
+    AguiEventConverter,
+)  # 向后兼容
+from agentrun.integration.langgraph.agent_converter import (
+    AgentRunConverter,
     convert,
     to_agui_events,
 )
 
 from .builtin import model, sandbox_toolset, toolset
 
 __all__ = [
-    "to_agui_events",
-    "convert",  # 兼容旧代码
+    "AgentRunConverter",
+    "AguiEventConverter",  # 向后兼容
+    "to_agui_events",  # 向后兼容
+    "convert",  # 向后兼容
     "model",
     "toolset",
     "sandbox_toolset",

agentrun/integration/langgraph/__init__.py

@@ -1,13 +1,13 @@
 """LangGraph 集成模块
 
-使用 to_agui_events 将 LangGraph 事件转换为 AG-UI 协议事件：
+使用 AgentRunConverter 将 LangGraph 事件转换为 AG-UI 协议事件：
 
-    >>> from agentrun.integration.langgraph import to_agui_events
+    >>> from agentrun.integration.langgraph import AgentRunConverter
     >>>
     >>> async def invoke_agent(request: AgentRequest):
-    ...     input_data = {"messages": [...]}
+    ...     converter = AgentRunConverter()
     ...     async for event in agent.astream_events(input_data, version="v2"):
-    ...         for item in to_agui_events(event):
+    ...         for item in converter.convert(event):
     ...             yield item
 
 支持多种调用方式：
@@ -16,12 +16,15 @@
 - agent.astream(input, stream_mode="updates") - 异步按节点输出
 """
 
-from .agent_converter import convert, to_agui_events
+from .agent_converter import AguiEventConverter  # 向后兼容
+from .agent_converter import AgentRunConverter, convert, to_agui_events
 from .builtin import model, sandbox_toolset, toolset
 
 __all__ = [
-    "to_agui_events",
-    "convert",  # 兼容旧代码
+    "AgentRunConverter",
+    "AguiEventConverter",  # 向后兼容
+    "to_agui_events",  # 向后兼容
+    "convert",  # 向后兼容
     "model",
     "toolset",
     "sandbox_toolset",

agentrun/integration/langgraph/agent_converter.py

@@ -498,11 +498,15 @@ def _convert_stream_values_event(
 
 def _convert_astream_events_event(
     event_dict: Dict[str, Any],
+    tool_call_id_map: Optional[Dict[int, str]] = None,
 ) -> Iterator[Union[AgentResult, str]]:
     """转换 astream_events 格式的单个事件
 
     Args:
         event_dict: 事件字典，格式为 {"event": "on_xxx", "data": {...}}
+        tool_call_id_map: 可选的 index -> tool_call_id 映射字典。
+                          在流式工具调用中，第一个 chunk 有 id，后续只有 index。
+                          此映射用于确保所有 chunk 使用一致的 tool_call_id。
 
     Yields:
         str (文本内容) 或 AgentResult (事件)
@@ -521,9 +525,34 @@ def _convert_astream_events_event(
 
             # 流式工具调用参数
             for tc in _extract_tool_call_chunks(chunk):
-                tc_id = tc.get("id") or str(tc.get("index", ""))
+                tc_index = tc.get("index")
+                tc_raw_id = tc.get("id")
                 tc_args = tc.get("args", "")
 
+                # 解析 tool_call_id：
+                # 1. 如果有 id 且非空，使用它并更新映射
+                # 2. 如果 id 为空但有 index，从映射中查找
+                # 3. 最后回退到使用 index 字符串
+                if tc_raw_id:
+                    tc_id = tc_raw_id
+                    # 更新映射（如果提供了映射字典）
+                    # 重要：即使这个 chunk 没有 args，也要更新映射，
+                    # 因为后续 chunk 可能只有 index 没有 id
+                    if tool_call_id_map is not None and tc_index is not None:
+                        tool_call_id_map[tc_index] = tc_id
+                elif tc_index is not None:
+                    # 从映射中查找，如果没有则使用 index
+                    if (
+                        tool_call_id_map is not None
+                        and tc_index in tool_call_id_map
+                    ):
+                        tc_id = tool_call_id_map[tc_index]
+                    else:
+                        tc_id = str(tc_index)
+                else:
+                    tc_id = ""
+
+                # 只有有 args 时才生成 TOOL_CALL_ARGS 事件
                 if tc_args and tc_id:
                     if isinstance(tc_args, (dict, list)):
                         tc_args = _safe_json_dumps(tc_args)
@@ -622,6 +651,7 @@ def _convert_astream_events_event(
 def to_agui_events(
     event: Union[Dict[str, Any], Any],
     messages_key: str = "messages",
+    tool_call_id_map: Optional[Dict[int, str]] = None,
 ) -> Iterator[Union[AgentResult, str]]:
     """将 LangGraph/LangChain 流式事件转换为 AG-UI 协议事件
 
@@ -635,12 +665,14 @@ def to_agui_events(
     Args:
         event: LangGraph/LangChain 流式事件（StreamEvent 对象或 Dict）
         messages_key: state 中消息列表的 key，默认 "messages"
+        tool_call_id_map: 可选的 index -> tool_call_id 映射字典，用于流式工具调用
+                          的 ID 一致性。如果提供，函数会自动更新此映射。
 
     Yields:
         str (文本内容) 或 AgentResult (AG-UI 事件)
 
     Example:
-        >>> # 使用 astream_events
+        >>> # 使用 astream_events（推荐使用 AguiEventConverter 类）
         >>> async for event in agent.astream_events(input, version="v2"):
         ...     for item in to_agui_events(event):
         ...         yield item
@@ -660,7 +692,7 @@ def to_agui_events(
     # 根据事件格式选择对应的转换器
     if _is_astream_events_format(event_dict):
         # astream_events 格式：{"event": "on_xxx", "data": {...}}
-        yield from _convert_astream_events_event(event_dict)
+        yield from _convert_astream_events_event(event_dict, tool_call_id_map)
 
     elif _is_stream_updates_format(event_dict):
         # stream/astream(stream_mode="updates") 格式：{node_name: state_update}
@@ -671,5 +703,55 @@ def to_agui_events(
         yield from _convert_stream_values_event(event_dict, messages_key)
 
 
+class AgentRunConverter:
+    """AgentRun 事件转换器
+
+    将 LangGraph/LangChain 流式事件转换为 AG-UI 协议事件。
+    此类维护必要的状态以确保流式工具调用的 tool_call_id 一致性。
+
+    在流式工具调用中，第一个 chunk 包含 id，后续 chunk 只有 index。
+    此类维护 index -> id 的映射，确保所有相关事件使用相同的 tool_call_id。
+
+    Example:
+        >>> from agentrun.integration.langchain import AgentRunConverter
+        >>>
+        >>> async def invoke_agent(request: AgentRequest):
+        ...     converter = AgentRunConverter()
+        ...     async for event in agent.astream_events(input, version="v2"):
+        ...         for item in converter.convert(event):
+        ...             yield item
+    """
+
+    def __init__(self):
+        self._tool_call_id_map: Dict[int, str] = {}
+
+    def convert(
+        self,
+        event: Union[Dict[str, Any], Any],
+        messages_key: str = "messages",
+    ) -> Iterator[Union[AgentResult, str]]:
+        """转换单个事件为 AG-UI 协议事件
+
+        Args:
+            event: LangGraph/LangChain 流式事件（StreamEvent 对象或 Dict）
+            messages_key: state 中消息列表的 key，默认 "messages"
+
+        Yields:
+            str (文本内容) 或 AgentResult (AG-UI 事件)
+        """
+        yield from to_agui_events(event, messages_key, self._tool_call_id_map)
+
+    def reset(self):
+        """重置状态，清空 tool_call_id 映射
+
+        在处理新的请求时，建议创建新的 AgentRunConverter 实例，
+        而不是复用旧实例并调用 reset。
+        """
+        self._tool_call_id_map.clear()
+
+
+# 保留向后兼容的别名
+AguiEventConverter = AgentRunConverter
+
 # 保留 convert 作为别名，兼容旧代码
 convert = to_agui_events