diff --git a/libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py b/libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py index 67a9a11dc..ad46692ec 100644 --- a/libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py +++ b/libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py @@ -1081,490 +1081,31 @@ def create_react_agent( print(chunk) ``` """ - if ( - config_schema := deprecated_kwargs.pop("config_schema", MISSING) - ) is not MISSING: - warn( - "`config_schema` is no longer supported. Use `context_schema` instead.", - category=LangGraphDeprecatedSinceV10, - ) - - if context_schema is not None: - context_schema = config_schema - - if version not in ("v1", "v2"): - raise ValueError( - f"Invalid version {version}. Supported versions are 'v1' and 'v2'." - ) - - if state_schema is not None: - required_keys = {"messages", "remaining_steps"} - if response_format is not None: - required_keys.add("structured_response") - - schema_keys = set(get_type_hints(state_schema)) - if missing_keys := required_keys - set(schema_keys): - raise ValueError(f"Missing required key(s) {missing_keys} in state_schema") - - if state_schema is None: - state_schema = ( - AgentStateWithStructuredResponse - if response_format is not None - else AgentState - ) - - llm_builtin_tools: list[dict] = [] - if isinstance(tools, ToolNode): - tool_classes = list(tools.tools_by_name.values()) - tool_node = tools - else: - llm_builtin_tools = [t for t in tools if isinstance(t, dict)] - tool_node = ToolNode([t for t in tools if not isinstance(t, dict)]) - tool_classes = list(tool_node.tools_by_name.values()) - - is_dynamic_model = not isinstance(model, (str, Runnable)) and callable(model) - is_async_dynamic_model = is_dynamic_model and inspect.iscoroutinefunction(model) - - tool_calling_enabled = len(tool_classes) > 0 - - if not is_dynamic_model: - if isinstance(model, str): - try: - from langchain.chat_models import ( # type: ignore[import-not-found] - init_chat_model, - ) - except ImportError: - raise ImportError( - "Please install langchain (`pip install langchain`) to " - "use ':' string syntax for `model` parameter." - ) - - model = cast(BaseChatModel, init_chat_model(model)) - - if ( - _should_bind_tools(model, tool_classes, num_builtin=len(llm_builtin_tools)) # type: ignore[arg-type] - and len(tool_classes + llm_builtin_tools) > 0 - ): - model = cast(BaseChatModel, model).bind_tools( - tool_classes + llm_builtin_tools # type: ignore[operator] - ) - - static_model: Optional[Runnable] = _get_prompt_runnable(prompt) | model # type: ignore[operator] - else: - # For dynamic models, we'll create the runnable at runtime - static_model = None - - # If any of the tools are configured to return_directly after running, - # our graph needs to check if these were called - should_return_direct = {t.name for t in tool_classes if t.return_direct} - - def _resolve_model( - state: StateSchema, runtime: Runtime[ContextT] - ) -> LanguageModelLike: - """Resolve the model to use, handling both static and dynamic models.""" - if is_dynamic_model: - return _get_prompt_runnable(prompt) | model(state, runtime) # type: ignore[operator] - else: - return static_model - - async def _aresolve_model( - state: StateSchema, runtime: Runtime[ContextT] - ) -> LanguageModelLike: - """Async resolve the model to use, handling both static and dynamic models.""" - if is_async_dynamic_model: - resolved_model = await model(state, runtime) # type: ignore[misc,operator] - return _get_prompt_runnable(prompt) | resolved_model - elif is_dynamic_model: - return _get_prompt_runnable(prompt) | model(state, runtime) # type: ignore[operator] - else: - return static_model - - def _are_more_steps_needed(state: StateSchema, response: BaseMessage) -> bool: - has_tool_calls = isinstance(response, AIMessage) and response.tool_calls - all_tools_return_direct = ( - all(call["name"] in should_return_direct for call in response.tool_calls) - if isinstance(response, AIMessage) - else False - ) - remaining_steps = _get_state_value(state, "remaining_steps", None) - is_last_step = _get_state_value(state, "is_last_step", False) - return ( - (remaining_steps is None and is_last_step and has_tool_calls) - or ( - remaining_steps is not None - and remaining_steps < 1 - and all_tools_return_direct - ) - or (remaining_steps is not None and remaining_steps < 2 and has_tool_calls) - ) - - def _get_model_input_state(state: StateSchema) -> StateSchema: - if pre_model_hook is not None: - messages = ( - _get_state_value(state, "llm_input_messages") - ) or _get_state_value(state, "messages") - error_msg = f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}" - else: - messages = _get_state_value(state, "messages") - error_msg = ( - f"Expected input to call_model to have 'messages' key, but got {state}" - ) - - if messages is None: - raise ValueError(error_msg) - - _validate_chat_history(messages) - # we're passing messages under `messages` key, as this is expected by the prompt - if isinstance(state_schema, type) and issubclass(state_schema, BaseModel): - state.messages = messages # type: ignore - else: - state["messages"] = messages # type: ignore - - return state - - # Define the function that calls the model - def call_model( - state: StateSchema, runtime: Runtime[ContextT], config: RunnableConfig - ) -> StateSchema: - if is_async_dynamic_model: - msg = ( - "Async model callable provided but agent invoked synchronously. " - "Use agent.ainvoke() or agent.astream(), or " - "provide a sync model callable." - ) - raise RuntimeError(msg) - - model_input = _get_model_input_state(state) - - if is_dynamic_model: - # Resolve dynamic model at runtime and apply prompt - dynamic_model = _resolve_model(state, runtime) - response = cast(AIMessage, dynamic_model.invoke(model_input, config)) # type: ignore[arg-type] - else: - response = cast(AIMessage, static_model.invoke(model_input, config)) # type: ignore[union-attr] - - # add agent name to the AIMessage - response.name = name - - if _are_more_steps_needed(state, response): - return { - "messages": [ - AIMessage( - id=response.id, - content="Sorry, need more steps to process this request.", - ) - ] - } - # We return a list, because this will get added to the existing list - return {"messages": [response]} - - async def acall_model( - state: StateSchema, runtime: Runtime[ContextT], config: RunnableConfig - ) -> StateSchema: - model_input = _get_model_input_state(state) - - if is_dynamic_model: - # Resolve dynamic model at runtime and apply prompt - # (supports both sync and async) - dynamic_model = await _aresolve_model(state, runtime) - response = cast(AIMessage, await dynamic_model.ainvoke(model_input, config)) # type: ignore[arg-type] - else: - response = cast(AIMessage, await static_model.ainvoke(model_input, config)) # type: ignore[union-attr] - - # add agent name to the AIMessage - response.name = name - if _are_more_steps_needed(state, response): - return { - "messages": [ - AIMessage( - id=response.id, - content="Sorry, need more steps to process this request.", - ) - ] - } - # We return a list, because this will get added to the existing list - return {"messages": [response]} - - input_schema: StateSchemaType - if pre_model_hook is not None: - # Dynamically create a schema that inherits from state_schema and adds 'llm_input_messages' - if isinstance(state_schema, type) and issubclass(state_schema, BaseModel): - # For Pydantic schemas - from pydantic import create_model - - input_schema = create_model( - "CallModelInputSchema", - llm_input_messages=(list[AnyMessage], ...), - __base__=state_schema, - ) - else: - # For TypedDict schemas - class CallModelInputSchema(state_schema): # type: ignore - llm_input_messages: list[AnyMessage] - - input_schema = CallModelInputSchema - else: - input_schema = state_schema - - def generate_structured_response( - state: StateSchema, runtime: Runtime[ContextT], config: RunnableConfig - ) -> StateSchema: - if is_async_dynamic_model: - msg = ( - "Async model callable provided but agent invoked synchronously. " - "Use agent.ainvoke() or agent.astream(), or provide a sync model callable." - ) - raise RuntimeError(msg) - - messages = _get_state_value(state, "messages") - structured_response_schema = response_format - if isinstance(response_format, tuple): - system_prompt, structured_response_schema = response_format - messages = [SystemMessage(content=system_prompt)] + list(messages) - - resolved_model = _resolve_model(state, runtime) - model_with_structured_output = _get_model( - resolved_model - ).with_structured_output( - cast(StructuredResponseSchema, structured_response_schema) - ) - response = model_with_structured_output.invoke(messages, config) - return {"structured_response": response} - - async def agenerate_structured_response( - state: StateSchema, runtime: Runtime[ContextT], config: RunnableConfig - ) -> StateSchema: - messages = _get_state_value(state, "messages") - structured_response_schema = response_format - if isinstance(response_format, tuple): - system_prompt, structured_response_schema = response_format - messages = [SystemMessage(content=system_prompt)] + list(messages) - - resolved_model = await _aresolve_model(state, runtime) - model_with_structured_output = _get_model( - resolved_model - ).with_structured_output( - cast(StructuredResponseSchema, structured_response_schema) - ) - response = await model_with_structured_output.ainvoke(messages, config) - return {"structured_response": response} - - if not tool_calling_enabled: - # Define a new graph - workflow = StateGraph(state_schema=state_schema, context_schema=context_schema) - workflow.add_node( - "agent", - RunnableCallable(call_model, acall_model), - input_schema=input_schema, - ) - if pre_model_hook is not None: - workflow.add_node("pre_model_hook", pre_model_hook) # type: ignore[arg-type] - workflow.add_edge("pre_model_hook", "agent") - entrypoint = "pre_model_hook" - else: - entrypoint = "agent" - - workflow.set_entry_point(entrypoint) - - if post_model_hook is not None: - workflow.add_node("post_model_hook", post_model_hook) # type: ignore[arg-type] - workflow.add_edge("agent", "post_model_hook") - - if response_format is not None: - workflow.add_node( - "generate_structured_response", - RunnableCallable( - generate_structured_response, - agenerate_structured_response, - ), - ) - if post_model_hook is not None: - workflow.add_edge("post_model_hook", "generate_structured_response") - else: - workflow.add_edge("agent", "generate_structured_response") - - return workflow.compile( - checkpointer=checkpointer, - store=store, - interrupt_before=interrupt_before, - interrupt_after=interrupt_after, - debug=debug, - name=name, - ) - - # Define the function that determines whether to continue or not - def should_continue(state: StateSchema) -> Union[str, list[Send]]: - messages = _get_state_value(state, "messages") - last_message = messages[-1] - # If there is no function call, then we finish - if not isinstance(last_message, AIMessage) or not last_message.tool_calls: - if post_model_hook is not None: - return "post_model_hook" - elif response_format is not None: - return "generate_structured_response" - else: - return END - # Otherwise if there is, we continue - else: - if version == "v1": - return "tools" - elif version == "v2": - if post_model_hook is not None: - return "post_model_hook" - return [ - Send( - "tools", - ToolCallWithContext( - __type="tool_call_with_context", - tool_call=tool_call, - state=state, - ), - ) - for tool_call in last_message.tool_calls - ] - - # Define a new graph - workflow = StateGraph( - state_schema=state_schema or AgentState, context_schema=context_schema - ) - - # Define the two nodes we will cycle between - workflow.add_node( - "agent", - RunnableCallable(call_model, acall_model), - input_schema=input_schema, - ) - workflow.add_node("tools", tool_node) - - # Optionally add a pre-model hook node that will be called - # every time before the "agent" (LLM-calling node) - if pre_model_hook is not None: - workflow.add_node("pre_model_hook", pre_model_hook) # type: ignore[arg-type] - workflow.add_edge("pre_model_hook", "agent") - entrypoint = "pre_model_hook" - else: - entrypoint = "agent" - - # Set the entrypoint as `agent` - # This means that this node is the first one called - workflow.set_entry_point(entrypoint) - - agent_paths = [] - post_model_hook_paths = [entrypoint, "tools"] - - # Add a post model hook node if post_model_hook is provided - if post_model_hook is not None: - workflow.add_node("post_model_hook", post_model_hook) # type: ignore[arg-type] - agent_paths.append("post_model_hook") - workflow.add_edge("agent", "post_model_hook") - else: - agent_paths.append("tools") - - # Add a structured output node if response_format is provided - if response_format is not None: - workflow.add_node( - "generate_structured_response", - RunnableCallable( - generate_structured_response, - agenerate_structured_response, - ), - ) - if post_model_hook is not None: - post_model_hook_paths.append("generate_structured_response") - else: - agent_paths.append("generate_structured_response") - else: - if post_model_hook is not None: - post_model_hook_paths.append(END) - else: - agent_paths.append(END) - - if post_model_hook is not None: - - def post_model_hook_router(state: StateSchema) -> Union[str, list[Send]]: - """Route to the next node after post_model_hook. - - Routes to one of: - * "tools": if there are pending tool calls without a corresponding message. - * "generate_structured_response": if no pending tool calls exist and response_format is specified. - * END: if no pending tool calls exist and no response_format is specified. - """ - - messages = _get_state_value(state, "messages") - tool_messages = [ - m.tool_call_id for m in messages if isinstance(m, ToolMessage) - ] - last_ai_message = next( - m for m in reversed(messages) if isinstance(m, AIMessage) - ) - pending_tool_calls = [ - c for c in last_ai_message.tool_calls if c["id"] not in tool_messages - ] - - if pending_tool_calls: - return [ - Send( - "tools", - ToolCallWithContext( - __type="tool_call_with_context", - tool_call=tool_call, - state=state, - ), - ) - for tool_call in pending_tool_calls - ] - elif isinstance(messages[-1], ToolMessage): - return entrypoint - elif response_format is not None: - return "generate_structured_response" - else: - return END - - workflow.add_conditional_edges( - "post_model_hook", - post_model_hook_router, # type: ignore[arg-type] - path_map=post_model_hook_paths, - ) - - workflow.add_conditional_edges( - "agent", - should_continue, # type: ignore[arg-type] - path_map=agent_paths, - ) - - def route_tool_responses(state: StateSchema) -> str: - for m in reversed(_get_state_value(state, "messages")): - if not isinstance(m, ToolMessage): - break - if m.name in should_return_direct: - return END - - # handle a case of parallel tool calls where - # the tool w/ `return_direct` was executed in a different `Send` - if isinstance(m, AIMessage) and m.tool_calls: - if any(call["name"] in should_return_direct for call in m.tool_calls): - return END - - return entrypoint - - if should_return_direct: - workflow.add_conditional_edges( - "tools", route_tool_responses, path_map=[entrypoint, END] - ) - else: - workflow.add_edge("tools", entrypoint) - - # Finally, we compile it! - # This compiles it into a LangChain Runnable, - # meaning you can use it as you would any other runnable - return workflow.compile( + # Parameter validation for deprecated kwargs + if deprecated_kwargs: + raise ValueError(f"Unexpected keyword arguments: {list(deprecated_kwargs.keys())}") + + # _AgentBuilder instantiation with all parameters + builder = _AgentBuilder( + model=model, + tools=tools, + prompt=prompt, + response_format=response_format, + pre_model_hook=pre_model_hook, + post_model_hook=post_model_hook, + state_schema=state_schema, + context_schema=context_schema, checkpointer=checkpointer, store=store, interrupt_before=interrupt_before, interrupt_after=interrupt_after, debug=debug, + version=version, name=name, ) + + # Return builder.build() + return builder.build() # Keep for backwards compatibility @@ -1587,3 +1128,4 @@ __all__ = [ +