Skip to content

fix: resolve message duplication in LangChain integration #40

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
OhYee merged 1 commit into from
Jan 21, 2026
Merged

fix: resolve message duplication in LangChain integration #40

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 13 additions & 7 deletions agentrun/integration/langgraph/agent_converter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,8 @@ def __init__(self) -> None:
# 用于在 on_tool_end 中查找对应的 tool_call_id
self._run_id_to_tool_call_id: Dict[str, str] = {}

self.has_on_chat_model_stream = False

def convert(
self,
event: Union[Dict[str, Any], Any],
Expand Down Expand Up @@ -436,8 +438,8 @@ def is_stream_values_format(event_dict: Dict[str, Any]) -> bool:
# 事件转换器（静态方法）
# =========================================================================

@staticmethod
def _convert_stream_updates_event(
Comment on lines 438 to 441
Copy link

Copilot AI Jan 21, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This section header still labels the following helpers as "静态方法" and the examples further down the file still show calling AgentRunConverter.to_agui_events(event) on the class, but to_agui_events has been converted to an instance method. To avoid confusing users, please update the comment and examples to describe and demonstrate instance-based usage (for example, via a converter instance) instead of static calls.

Copilot uses AI. Check for mistakes.
self,
event_dict: Dict[str, Any],
messages_key: str = "messages",
) -> Iterator[Union[AgentResult, str]]:
Expand Down Expand Up @@ -525,8 +527,8 @@ def _convert_stream_updates_event(
},
)

@staticmethod
def _convert_stream_values_event(
self,
event_dict: Dict[str, Any],
messages_key: str = "messages",
) -> Iterator[Union[AgentResult, str]]:
Expand Down Expand Up @@ -595,8 +597,8 @@ def _convert_stream_values_event(
},
)

@staticmethod
def _convert_astream_events_event(
self,
event_dict: Dict[str, Any],
tool_call_id_map: Optional[Dict[int, str]] = None,
tool_call_started_set: Optional[set] = None,
Expand Down Expand Up @@ -625,6 +627,9 @@ def _convert_astream_events_event(

# 1. LangGraph 格式: on_chat_model_stream
if event_type == "on_chat_model_stream":

self.has_on_chat_model_stream = True

chunk = data.get("chunk")
if chunk:
# 文本内容
Expand Down Expand Up @@ -722,6 +727,7 @@ def _convert_astream_events_event(
elif (
event_type == "on_chain_stream"
and event_dict.get("name") == "model"
and not self.has_on_chat_model_stream
):
chunk_data = data.get("chunk", {})
if isinstance(chunk_data, dict):
Expand Down Expand Up @@ -997,8 +1003,8 @@ def _convert_astream_events_event(
# 主要 API（静态方法）
# =========================================================================

@staticmethod
def to_agui_events(
self,
event: Union[Dict[str, Any], Any],
messages_key: str = "messages",
tool_call_id_map: Optional[Dict[int, str]] = None,
Expand Down Expand Up @@ -1052,7 +1058,7 @@ def to_agui_events(
# 根据事件格式选择对应的转换器
if AgentRunConverter.is_astream_events_format(event_dict):
# astream_events 格式：{"event": "on_xxx", "data": {...}}
yield from AgentRunConverter._convert_astream_events_event(
yield from self._convert_astream_events_event(
event_dict,
tool_call_id_map,
tool_call_started_set,
Expand All @@ -1062,12 +1068,12 @@ def to_agui_events(

elif AgentRunConverter.is_stream_updates_format(event_dict):
# stream/astream(stream_mode="updates") 格式：{node_name: state_update}
yield from AgentRunConverter._convert_stream_updates_event(
yield from self._convert_stream_updates_event(
event_dict, messages_key
)

elif AgentRunConverter.is_stream_values_format(event_dict):
# stream/astream(stream_mode="values") 格式：完整 state dict
yield from AgentRunConverter._convert_stream_values_event(
yield from self._convert_stream_values_event(
event_dict, messages_key
)
Loading
Loading