jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

fix: Streaming for anthropic models in Agents (#7871)

Browse source 
* fix streaming for anthropic models and optimise the _extract_output_text

* Update src/backend/base/langflow/base/agents/events.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* fix: improve handling of AIMessageChunk state and output text extraction

* fix: change ValueError to TypeError in output text extraction

* fix: ensure output text is appended only if it is valid and agent message text is a string

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Gabriel Luiz Freitas Almeida <gabriel@langflow.org>
This commit is contained in:
Edwin Jose 2025-05-02 10:49:44 -04:00 • committed by GitHub
commit 2fa6d868a1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 23 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

32
src/backend/base/langflow/base/agents/events.py
View file

@ -81,13 +81,25 @@ async def handle_on_chain_start(
def _extract_output_text(output: str | list) -> str:
if isinstance(output, str):
text = output
elif isinstance(output, list) and len(output) == 1 and isinstance(output[0], dict) and "text" in output[0]:
text = output[0]["text"]
else:
return output
if isinstance(output, list) and len(output) == 0:
return ""
if not isinstance(output, list) or len(output) != 1:
msg = f"Output is not a string or list of dictionaries with 'text' key: {output}"
raise ValueError(msg)
return text
raise TypeError(msg)
item = output[0]
if isinstance(item, str):
return item
if isinstance(item, dict):
if "text" in item:
return item["text"]
# If the item's type is "tool_use", return an empty string.
# This likely indicates that "tool_use" outputs are not meant to be displayed as text.
if item.get("type") == "tool_use":
return ""
msg = f"Output is not a string or list of dictionaries with 'text' key: {output}"
raise TypeError(msg)
async def handle_on_chain_end(
@ -214,9 +226,11 @@ async def handle_on_chain_stream(
agent_message = await send_message_method(message=agent_message)
start_time = perf_counter()
elif isinstance(data_chunk, AIMessageChunk):
agent_message.text += data_chunk.content
agent_message.properties.state = "complete"
agent_message = await send_message_method(message=agent_message)
output_text = _extract_output_text(data_chunk.content)
if output_text and isinstance(agent_message.text, str):
agent_message.text += output_text
agent_message.properties.state = "partial"
agent_message = await send_message_method(message=agent_message)
start_time = perf_counter()
return agent_message, start_time