@@ -48,8 +48,8 @@ async def convert_langgraph_to_agentex_events( |
48 | 48 | Supports both regular models (chunk.content is a str) and reasoning models |
49 | 49 | like gpt-5/o1/o3 (chunk.content is a list of typed content blocks). |
50 | 50 |
|
51 | | - AGX1-377 note: LangGraph emits tool requests as ``StreamTaskMessageFull`` (from |
52 | | - "updates" events), NOT Start+Delta+Done like pydantic-ai. No coalesce_tool_requests |
| 51 | + LangGraph emits tool requests as ``StreamTaskMessageFull`` (from "updates" |
| 52 | + events), NOT Start+Delta+Done like pydantic-ai. No coalesce_tool_requests |
53 | 53 | option is needed for LangGraph. |
54 | 54 |
|
55 | 55 | Args: |
@@ -271,3 +271,82 @@ async def convert_langgraph_to_agentex_events( |
271 | 271 | yield StreamTaskMessageDone(type="done", index=message_index) |
272 | 272 | if reasoning_streaming: |
273 | 273 | yield StreamTaskMessageDone(type="done", index=message_index) |
| 274 | + |
| 275 | + |
| 276 | +async def emit_langgraph_messages(messages: list[Any], task_id: str) -> str: |
| 277 | + """Create Agentex messages for a list of LangGraph messages. |
| 278 | +
|
| 279 | + This is the non-streaming counterpart to ``stream_langgraph_events``. Use it |
| 280 | + when you run a LangGraph graph with ``ainvoke`` (for example a Temporal-backed |
| 281 | + agent using the LangGraph plugin, where streaming deltas aren't available) and |
| 282 | + want to surface the resulting messages to the Agentex UI after the fact. |
| 283 | +
|
| 284 | + It maps LangGraph/LangChain message objects to Agentex content types: |
| 285 | +
|
| 286 | + - ``AIMessage`` tool calls -> ``ToolRequestContent`` (one per call) |
| 287 | + - ``AIMessage`` text content -> ``TextContent`` |
| 288 | + - ``ToolMessage`` -> ``ToolResponseContent`` |
| 289 | +
|
| 290 | + Pass only the messages produced this turn (e.g. ``messages[already_emitted:]``) |
| 291 | + so each message is surfaced exactly once across a multi-turn conversation. |
| 292 | +
|
| 293 | + Args: |
| 294 | + messages: LangGraph/LangChain message objects to surface — typically |
| 295 | + the new messages a turn produced. |
| 296 | + task_id: The Agentex task to create messages on. |
| 297 | +
|
| 298 | + Returns: |
| 299 | + The last assistant text emitted (useful as a span/turn output), or "". |
| 300 | + """ |
| 301 | + # Lazy imports so langchain isn't required at module load time. |
| 302 | + from langchain_core.messages import AIMessage, ToolMessage |
| 303 | + |
| 304 | + from agentex.lib import adk |
| 305 | + from agentex.types.text_content import TextContent |
| 306 | + from agentex.types.tool_request_content import ToolRequestContent |
| 307 | + from agentex.types.tool_response_content import ToolResponseContent |
| 308 | + |
| 309 | + final_text = "" |
| 310 | + for message in messages: |
| 311 | + if isinstance(message, AIMessage): |
| 312 | + for tool_call in message.tool_calls or []: |
| 313 | + await adk.messages.create( |
| 314 | + task_id=task_id, |
| 315 | + content=ToolRequestContent( |
| 316 | + author="agent", |
| 317 | + tool_call_id=tool_call["id"], |
| 318 | + name=tool_call["name"], |
| 319 | + arguments=tool_call["args"], |
| 320 | + ), |
| 321 | + ) |
| 322 | + # ``content`` may be a plain string (OpenAI) or a list of content |
| 323 | + # blocks (Anthropic/Claude via LangChain, e.g. |
| 324 | + # ``[{"type": "text", "text": "..."}]``). Extract and join the text |
| 325 | + # so the response is visible regardless of the underlying model. |
| 326 | + if isinstance(message.content, str): |
| 327 | + text = message.content |
| 328 | + else: |
| 329 | + text = "".join( |
| 330 | + block.get("text", "") if isinstance(block, dict) else str(block) |
| 331 | + for block in message.content |
| 332 | + if not isinstance(block, dict) or block.get("type") == "text" |
| 333 | + ) |
| 334 | + if text: |
| 335 | + final_text = text |
| 336 | + await adk.messages.create( |
| 337 | + task_id=task_id, |
| 338 | + content=TextContent(author="agent", content=text, format="markdown"), |
| 339 | + ) |
| 340 | + elif isinstance(message, ToolMessage): |
| 341 | + await adk.messages.create( |
| 342 | + task_id=task_id, |
| 343 | + content=ToolResponseContent( |
| 344 | + author="agent", |
| 345 | + tool_call_id=message.tool_call_id, |
| 346 | + name=message.name or "unknown", |
| 347 | + content=message.content |
| 348 | + if isinstance(message.content, str) |
| 349 | + else str(message.content), |
| 350 | + ), |
| 351 | + ) |
| 352 | + return final_text |