refactor: Remove unused stream_task method from LangGraphAgentAdapter

- Deleted the `stream_task` method from LangGraphAgentAdapter to streamline the code and eliminate unnecessary complexity.
- This change enhances maintainability by focusing on essential functionalities within the agent adapter.
This commit is contained in:
lorenzejay
2025-04-16 16:00:42 -07:00
parent d02de6ea38
commit 51795cf1e7

View File

@@ -198,71 +198,6 @@ class LangGraphAgentAdapter(BaseAgentAdapter):
)
raise
async def stream_task(
self,
task: Any,
context: Optional[str] = None,
tools: Optional[List[BaseTool]] = None,
) -> AsyncIterable[Dict[str, Any]]:
"""Stream the execution of a task."""
self.create_agent_executor(tools)
try:
task_prompt = task.prompt() if hasattr(task, "prompt") else str(task)
if context:
task_prompt = self.i18n.slice("task_with_context").format(
task=task_prompt, context=context
)
# Set up a session ID for this task
session_id = f"task_{id(task)}"
# Configure the invocation
config = {"configurable": {"thread_id": session_id}}
# Stream the execution
inputs = {"messages": [("user", task_prompt)]}
for item in self._graph.stream(inputs, config, stream_mode="values"):
message = item.get("messages", [])[-1] if "messages" in item else None
if (
message is not None
and hasattr(message, "tool_calls")
and getattr(message, "tool_calls", None)
):
tool_calls = getattr(message, "tool_calls", [])
if tool_calls and len(tool_calls) > 0:
yield {
"is_task_complete": False,
"require_user_input": False,
"content": f"Using tool: {tool_calls[0].name}",
}
elif isinstance(message, ToolMessage):
content = getattr(message, "content", "Tool execution complete")
yield {
"is_task_complete": False,
"require_user_input": False,
"content": f"Tool result: {content[:50]}...",
}
elif message is not None:
# Final response or intermediary thinking
content = getattr(message, "content", str(message))
yield {
"is_task_complete": True,
"require_user_input": False,
"content": content,
}
except Exception as e:
self._logger.log("error", f"Error streaming LangGraph task: {str(e)}")
yield {
"is_task_complete": True,
"require_user_input": False,
"content": f"Error: {str(e)}",
}
def create_agent_executor(self, tools: Optional[List[BaseTool]] = None) -> None:
"""Configure the LangGraph agent for execution."""
self.configure_tools(tools)